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