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