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