Real Estate Expert

Real Estate Expert

Expert guidance for real estate systems, property management, Multiple Listing Service (MLS) integration, customer relationship management, virtual tours, and market analysis.

Core Concepts

Real Estate Systems

• Multiple Listing Service (MLS) integration
• Property Management Systems (PMS)
• Customer Relationship Management (CRM)
• Transaction management
• Document management
• Lease management
• Maintenance tracking

PropTech Solutions

• Virtual tours and 3D walkthroughs
• AI-powered property valuation
• Digital signatures and e-closing
• Smart home integration
• IoT sensors for properties
• Blockchain for title management
• Augmented reality for staging

Standards and Regulations

• RESO (Real Estate Standards Organization)
• Fair Housing Act compliance
• RESPA (Real Estate Settlement Procedures Act)
• Data privacy (GDPR, CCPA)
• ADA compliance for websites
• NAR Code of Ethics

Property Listing System

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from typing import List, Optional
from enum import Enum

class PropertyType(Enum):
    SINGLE_FAMILY = "single_family"
    CONDO = "condo"
    TOWNHOUSE = "townhouse"
    MULTI_FAMILY = "multi_family"
    LAND = "land"
    COMMERCIAL = "commercial"

class ListingStatus(Enum):
    ACTIVE = "active"
    PENDING = "pending"
    SOLD = "sold"
    WITHDRAWN = "withdrawn"
    EXPIRED = "expired"

@dataclass
class Property:
    """Property information"""
    property_id: str
    mls_number: str
    property_type: PropertyType
    address: dict
    listing_price: Decimal
    bedrooms: int
    bathrooms: float
    square_feet: int
    lot_size: float  # acres
    year_built: int
    description: str
    features: List[str]
    photos: List[str]
    status: ListingStatus
    listing_date: datetime
    listing_agent_id: str
    coordinates: tuple  # (latitude, longitude)

@dataclass
class ShowingRequest:
    """Property showing request"""
    showing_id: str
    property_id: str
    buyer_agent_id: str
    buyer_name: str
    requested_date: datetime
    duration_minutes: int
    status: str  # 'pending', 'confirmed', 'cancelled'
    notes: str

class PropertyListingSystem:
    """Real estate listing management system"""

    def __init__(self):
        self.properties = {}
        self.showings = []
        self.saved_searches = {}

    def create_listing(self,
                      property_data: dict,
                      agent_id: str) -> Property:
        """Create new property listing"""
        property_id = self._generate_property_id()
        mls_number = self._generate_mls_number()

        property = Property(
            property_id=property_id,
            mls_number=mls_number,
            property_type=PropertyType(property_data['property_type']),
            address=property_data['address'],
            listing_price=Decimal(str(property_data['price'])),
            bedrooms=property_data['bedrooms'],
            bathrooms=property_data['bathrooms'],
            square_feet=property_data['square_feet'],
            lot_size=property_data.get('lot_size', 0),
            year_built=property_data['year_built'],
            description=property_data['description'],
            features=property_data.get('features', []),
            photos=property_data.get('photos', []),
            status=ListingStatus.ACTIVE,
            listing_date=datetime.now(),
            listing_agent_id=agent_id,
            coordinates=property_data.get('coordinates', (0, 0))
        )

        self.properties[property_id] = property

        # Notify matching saved searches
        self._notify_saved_searches(property)

        return property

    def search_properties(self, criteria: dict) -> List[Property]:
        """Search properties based on criteria"""
        results = []

        for property in self.properties.values():
            if property.status != ListingStatus.ACTIVE:
                continue

            # Price range
            if 'min_price' in criteria:
                if property.listing_price < Decimal(str(criteria['min_price'])):
                    continue

            if 'max_price' in criteria:
                if property.listing_price > Decimal(str(criteria['max_price'])):
                    continue

            # Bedrooms
            if 'min_bedrooms' in criteria:
                if property.bedrooms < criteria['min_bedrooms']:
                    continue

            # Bathrooms
            if 'min_bathrooms' in criteria:
                if property.bathrooms < criteria['min_bathrooms']:
                    continue

            # Square footage
            if 'min_sqft' in criteria:
                if property.square_feet < criteria['min_sqft']:
                    continue

            # Property type
            if 'property_type' in criteria:
                if property.property_type.value != criteria['property_type']:
                    continue

            # Location-based search (within radius)
            if 'location' in criteria and 'radius_miles' in criteria:
                distance = self._calculate_distance(
                    property.coordinates,
                    criteria['location']
                )
                if distance > criteria['radius_miles']:
                    continue

            results.append(property)

        # Sort by price or other criteria
        if criteria.get('sort_by') == 'price_asc':
            results.sort(key=lambda p: p.listing_price)
        elif criteria.get('sort_by') == 'price_desc':
            results.sort(key=lambda p: p.listing_price, reverse=True)
        elif criteria.get('sort_by') == 'newest':
            results.sort(key=lambda p: p.listing_date, reverse=True)

        return results

    def schedule_showing(self,
                        property_id: str,
                        buyer_agent_id: str,
                        buyer_name: str,
                        requested_date: datetime) -> dict:
        """Schedule property showing"""
        property = self.properties.get(property_id)
        if not property:
            return {'error': 'Property not found'}

        if property.status != ListingStatus.ACTIVE:
            return {'error': 'Property not available for showings'}

        # Check availability
        conflicts = self._check_showing_conflicts(property_id, requested_date)
        if conflicts:
            return {
                'error': 'Time slot not available',
                'conflicts': conflicts
            }

        showing = ShowingRequest(
            showing_id=self._generate_showing_id(),
            property_id=property_id,
            buyer_agent_id=buyer_agent_id,
            buyer_name=buyer_name,
            requested_date=requested_date,
            duration_minutes=30,
            status='pending',
            notes=''
        )

        self.showings.append(showing)

        # Notify listing agent
        self._notify_listing_agent(property.listing_agent_id, showing)

        return {
            'success': True,
            'showing_id': showing.showing_id,
            'status': 'pending_confirmation'
        }

    def calculate_price_per_sqft(self, property: Property) -> Decimal:
        """Calculate price per square foot"""
        if property.square_feet == 0:
            return Decimal('0')

        price_per_sqft = property.listing_price / property.square_feet
        return price_per_sqft.quantize(Decimal('0.01'))

    def generate_cma(self,
                    subject_property: Property,
                    radius_miles: float = 1.0) -> dict:
        """Generate Comparative Market Analysis (CMA)"""
        # Find comparable properties
        comparables = []

        for property in self.properties.values():
            # Skip the subject property
            if property.property_id == subject_property.property_id:
                continue

            # Similar property type
            if property.property_type != subject_property.property_type:
                continue

            # Recently sold (last 6 months)
            if property.status != ListingStatus.SOLD:
                continue

            days_since_sale = (datetime.now() - property.listing_date).days
            if days_since_sale > 180:
                continue

            # Within radius
            distance = self._calculate_distance(
                subject_property.coordinates,
                property.coordinates
            )
            if distance > radius_miles:
                continue

            # Similar size (within 20%)
            size_diff = abs(property.square_feet - subject_property.square_feet)
            size_diff_pct = size_diff / subject_property.square_feet
            if size_diff_pct > 0.2:
                continue

            # Similar bedrooms
            if abs(property.bedrooms - subject_property.bedrooms) > 1:
                continue

            comparables.append(property)

        if not comparables:
            return {'error': 'No comparable properties found'}

        # Calculate statistics
        prices = [float(p.listing_price) for p in comparables]
        price_per_sqft_values = [
            float(self.calculate_price_per_sqft(p)) for p in comparables
        ]

        avg_price = sum(prices) / len(prices)
        avg_price_per_sqft = sum(price_per_sqft_values) / len(price_per_sqft_values)

        # Estimate subject property value
        estimated_value = avg_price_per_sqft * subject_property.square_feet

        return {
            'subject_property_id': subject_property.property_id,
            'comparable_count': len(comparables),
            'comparables': [
                {
                    'property_id': p.property_id,
                    'address': p.address,
                    'price': float(p.listing_price),
                    'square_feet': p.square_feet,
                    'price_per_sqft': float(self.calculate_price_per_sqft(p))
                }
                for p in comparables[:5]  # Top 5 comparables
            ],
            'market_statistics': {
                'average_price': avg_price,
                'average_price_per_sqft': avg_price_per_sqft,
                'min_price': min(prices),
                'max_price': max(prices)
            },
            'estimated_value': estimated_value,
            'suggested_listing_price': estimated_value * 0.98  # Slightly below estimate
        }

    def save_search(self, user_id: str, search_criteria: dict) -> str:
        """Save search criteria for notifications"""
        search_id = self._generate_search_id()

        self.saved_searches[search_id] = {
            'user_id': user_id,
            'criteria': search_criteria,
            'created_at': datetime.now(),
            'active': True
        }

        return search_id

    def _calculate_distance(self, coord1: tuple, coord2: tuple) -> float:
        """Calculate distance between two coordinates (miles)"""
        from math import radians, sin, cos, sqrt, atan2

        lat1, lon1 = radians(coord1[0]), radians(coord1[1])
        lat2, lon2 = radians(coord2[0]), radians(coord2[1])

        dlat = lat2 - lat1
        dlon = lon2 - lon1

        a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**2
        c = 2 * atan2(sqrt(a), sqrt(1-a))

        radius_miles = 3959  # Earth's radius in miles
        distance = radius_miles * c

        return distance

    def _check_showing_conflicts(self,
                                property_id: str,
                                requested_date: datetime) -> List[dict]:
        """Check for scheduling conflicts"""
        conflicts = []

        for showing in self.showings:
            if showing.property_id != property_id:
                continue

            if showing.status == 'cancelled':
                continue

            # Check for time overlap (within 1 hour)
            time_diff = abs((showing.requested_date - requested_date).total_seconds() / 3600)
            if time_diff < 1:
                conflicts.append({
                    'showing_id': showing.showing_id,
                    'time': showing.requested_date.isoformat()
                })

        return conflicts

    def _notify_saved_searches(self, property: Property):
        """Notify users with matching saved searches"""
        # Implementation would check saved searches and send notifications
        pass

    def _notify_listing_agent(self, agent_id: str, showing: ShowingRequest):
        """Notify listing agent of showing request"""
        # Implementation would send email/SMS notification
        pass

    def _generate_property_id(self) -> str:
        import uuid
        return f"PROP-{uuid.uuid4().hex[:8].upper()}"

    def _generate_mls_number(self) -> str:
        import uuid
        return f"MLS-{uuid.uuid4().hex[:10].upper()}"

    def _generate_showing_id(self) -> str:
        import uuid
        return f"SHOW-{uuid.uuid4().hex[:8].upper()}"

    def _generate_search_id(self) -> str:
        import uuid
        return f"SEARCH-{uuid.uuid4().hex[:8].upper()}"

Property Valuation and Analytics

import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.preprocessing import StandardScaler

class PropertyValuationSystem:
    """AI-powered property valuation"""

    def __init__(self):
        self.model = GradientBoostingRegressor(n_estimators=100)
        self.scaler = StandardScaler()
        self.trained = False

    def train_model(self, training_data: List[dict]):
        """Train valuation model on historical data"""
        features = []
        prices = []

        for property_data in training_data:
            feature_vector = self._extract_features(property_data)
            features.append(feature_vector)
            prices.append(property_data['sold_price'])

        X = np.array(features)
        y = np.array(prices)

        # Scale features
        X_scaled = self.scaler.fit_transform(X)

        # Train model
        self.model.fit(X_scaled, y)
        self.trained = True

    def estimate_value(self, property_data: dict) -> dict:
        """Estimate property value"""
        if not self.trained:
            return {'error': 'Model not trained'}

        features = self._extract_features(property_data)
        features_scaled = self.scaler.transform([features])

        estimated_value = self.model.predict(features_scaled)[0]

        # Calculate confidence interval (simplified)
        confidence_range = estimated_value * 0.1  # ±10%

        return {
            'estimated_value': estimated_value,
            'confidence_interval': {
                'lower': estimated_value - confidence_range,
                'upper': estimated_value + confidence_range
            },
            'price_per_sqft': estimated_value / property_data['square_feet']
        }

    def _extract_features(self, property_data: dict) -> List[float]:
        """Extract features for valuation model"""
        return [
            property_data['square_feet'],
            property_data['bedrooms'],
            property_data['bathrooms'],
            property_data['lot_size'],
            property_data['year_built'],
            property_data.get('garage_spaces', 0),
            property_data.get('stories', 1),
            1 if property_data.get('has_pool', False) else 0,
            1 if property_data.get('has_fireplace', False) else 0,
            property_data.get('neighborhood_score', 50)  # 0-100 scale
        ]

class MarketAnalytics:
    """Real estate market analytics"""

    def calculate_market_trends(self, sales_data: List[dict]) -> dict:
        """Calculate market trends and statistics"""
        if not sales_data:
            return {'error': 'No sales data available'}

        # Calculate metrics
        prices = [s['price'] for s in sales_data]
        days_on_market = [s['days_on_market'] for s in sales_data]

        median_price = np.median(prices)
        avg_price = np.mean(prices)
        avg_days_on_market = np.mean(days_on_market)

        # Calculate price trends (compare recent vs older data)
        recent_data = sales_data[-30:]  # Last 30 sales
        older_data = sales_data[-60:-30]  # Previous 30 sales

        if len(recent_data) > 0 and len(older_data) > 0:
            recent_avg = np.mean([s['price'] for s in recent_data])
            older_avg = np.mean([s['price'] for s in older_data])
            price_change = ((recent_avg - older_avg) / older_avg) * 100
        else:
            price_change = 0

        # Market health indicator
        if avg_days_on_market < 30:
            market_health = "Hot"
        elif avg_days_on_market < 60:
            market_health = "Balanced"
        else:
            market_health = "Slow"

        return {
            'median_price': median_price,
            'average_price': avg_price,
            'average_days_on_market': avg_days_on_market,
            'price_trend_percentage': price_change,
            'market_health': market_health,
            'total_sales': len(sales_data)
        }

    def calculate_inventory_metrics(self, active_listings: List[Property]) -> dict:
        """Calculate inventory and absorption metrics"""
        total_listings = len(active_listings)

        # Calculate average price
        avg_price = np.mean([float(p.listing_price) for p in active_listings])

        # Calculate months of inventory (simplified)
        # Would need sales velocity for accurate calculation
        months_of_inventory = 6.0  # Placeholder

        return {
            'total_active_listings': total_listings,
            'average_listing_price': avg_price,
            'months_of_inventory': months_of_inventory,
            'market_condition': 'Balanced' if 4 <= months_of_inventory <= 6 else
                              'Seller' if months_of_inventory < 4 else 'Buyer'
        }

Lease Management

@dataclass
class Lease:
    """Rental lease agreement"""
    lease_id: str
    property_id: str
    tenant_name: str
    tenant_contact: dict
    start_date: datetime
    end_date: datetime
    monthly_rent: Decimal
    security_deposit: Decimal
    status: str  # 'active', 'expired', 'terminated'
    auto_renew: bool

@dataclass
class MaintenanceRequest:
    """Maintenance request for property"""
    request_id: str
    property_id: str
    tenant_name: str
    category: str  # 'plumbing', 'electrical', 'hvac', etc.
    priority: str  # 'low', 'medium', 'high', 'emergency'
    description: str
    submitted_date: datetime
    status: str  # 'open', 'in_progress', 'completed'
    assigned_to: Optional[str]

class PropertyManagementSystem:
    """Property management for landlords and property managers"""

    def __init__(self):
        self.leases = {}
        self.maintenance_requests = []
        self.rent_payments = []

    def create_lease(self, lease_data: dict) -> Lease:
        """Create new lease agreement"""
        lease_id = self._generate_lease_id()

        lease = Lease(
            lease_id=lease_id,
            property_id=lease_data['property_id'],
            tenant_name=lease_data['tenant_name'],
            tenant_contact=lease_data['tenant_contact'],
            start_date=lease_data['start_date'],
            end_date=lease_data['end_date'],
            monthly_rent=Decimal(str(lease_data['monthly_rent'])),
            security_deposit=Decimal(str(lease_data['security_deposit'])),
            status='active',
            auto_renew=lease_data.get('auto_renew', False)
        )

        self.leases[lease_id] = lease

        # Schedule rent payment reminders
        self._schedule_rent_reminders(lease)

        return lease

    def record_rent_payment(self,
                           lease_id: str,
                           amount: Decimal,
                           payment_date: datetime,
                           payment_method: str) -> dict:
        """Record rent payment"""
        lease = self.leases.get(lease_id)
        if not lease:
            return {'error': 'Lease not found'}

        payment = {
            'payment_id': self._generate_payment_id(),
            'lease_id': lease_id,
            'amount': amount,
            'payment_date': payment_date,
            'payment_method': payment_method,
            'for_month': payment_date.strftime('%Y-%m')
        }

        self.rent_payments.append(payment)

        # Check if payment is late
        expected_date = datetime(payment_date.year, payment_date.month, 1)
        days_late = (payment_date - expected_date).days

        return {
            'success': True,
            'payment_id': payment['payment_id'],
            'days_late': max(0, days_late),
            'late_fee': self._calculate_late_fee(lease, days_late)
        }

    def submit_maintenance_request(self, request_data: dict) -> MaintenanceRequest:
        """Submit maintenance request"""
        request = MaintenanceRequest(
            request_id=self._generate_request_id(),
            property_id=request_data['property_id'],
            tenant_name=request_data['tenant_name'],
            category=request_data['category'],
            priority=request_data.get('priority', 'medium'),
            description=request_data['description'],
            submitted_date=datetime.now(),
            status='open',
            assigned_to=None
        )

        self.maintenance_requests.append(request)

        # Auto-assign emergency requests
        if request.priority == 'emergency':
            self._assign_emergency_maintenance(request)

        return request

    def check_lease_expiration(self) -> List[dict]:
        """Check for expiring leases"""
        expiring_soon = []
        current_date = datetime.now()

        for lease in self.leases.values():
            if lease.status != 'active':
                continue

            days_until_expiration = (lease.end_date - current_date).days

            if 0 < days_until_expiration <= 60:
                expiring_soon.append({
                    'lease_id': lease.lease_id,
                    'property_id': lease.property_id,
                    'tenant_name': lease.tenant_name,
                    'end_date': lease.end_date.isoformat(),
                    'days_remaining': days_until_expiration,
                    'auto_renew': lease.auto_renew
                })

        return expiring_soon

    def _calculate_late_fee(self, lease: Lease, days_late: int) -> Decimal:
        """Calculate late fee for rent payment"""
        if days_late <= 5:  # Grace period
            return Decimal('0')

        # $50 flat fee + $5 per day after grace period
        late_fee = Decimal('50') + (Decimal('5') * (days_late - 5))
        return late_fee

    def _schedule_rent_reminders(self, lease: Lease):
        """Schedule monthly rent payment reminders"""
        # Implementation would schedule reminder emails/notifications
        pass

    def _assign_emergency_maintenance(self, request: MaintenanceRequest):
        """Auto-assign emergency maintenance requests"""
        # Implementation would assign to on-call maintenance staff
        pass

    def _generate_lease_id(self) -> str:
        import uuid
        return f"LEASE-{uuid.uuid4().hex[:8].upper()}"

    def _generate_payment_id(self) -> str:
        import uuid
        return f"PAY-{uuid.uuid4().hex[:8].upper()}"

    def _generate_request_id(self) -> str:
        import uuid
        return f"MAINT-{uuid.uuid4().hex[:8].upper()}"

Best Practices

Listing Management

• Use high-quality professional photos
• Write compelling property descriptions
• Include virtual tours and 3D walkthroughs
• Update listings immediately when status changes
• Respond to inquiries within 1 hour
• Maintain accurate MLS data
• Use targeted marketing campaigns

Property Valuation

• Use multiple valuation methods (CMA, AVM, appraisal)
• Consider local market conditions
• Account for property condition and upgrades
• Review comparable sales regularly
• Factor in seasonal trends
• Include neighborhood analysis
• Document valuation methodology

Lease Management

• Use standardized lease templates
• Conduct thorough tenant screening
• Document property condition (move-in/move-out)
• Maintain security deposit in separate account
• Schedule regular property inspections
• Respond to maintenance requests promptly
• Maintain clear communication with tenants

Compliance

• Follow Fair Housing Act requirements
• Maintain proper licensing
• Use compliant lease agreements
• Protect tenant privacy
• Follow eviction procedures properly
• Maintain insurance coverage
• Keep accurate financial records

Anti-Patterns

❌ Poor quality listing photos ❌ Inaccurate property information ❌ Slow response to inquiries ❌ No virtual tour options ❌ Ignoring online reviews ❌ Manual document management ❌ No tenant screening process ❌ Poor maintenance tracking ❌ Inadequate insurance coverage

Resources

• NAR (National Association of Realtors): https://www.nar.realtor/
• RESO Standards: https://www.reso.org/
• Zillow API: https://www.zillow.com/howto/api/
• Realtor.com API: https://www.realtor.com/
• CoreLogic: https://www.corelogic.com/
• Redfin Data: https://www.redfin.com/
• Fair Housing Act: https://www.hud.gov/fairhousing