Rental Leads / Rental Comps

# parse_listings.py - Scrape & extract rental listings with realistic fixture from bs4 import BeautifulSoup import json SAMPLE_HTML_FIXTURE = """ """ def extract_listings_from_html(html_text: str) -> list: soup = BeautifulSoup(html_text, "html.parser") extracted = [] # 1. Parse structured JSON-LD schemas scripts = soup.find_all("script", type="application/ld+json") for s in scripts: try: data = json.loads(s.string) if isinstance(data, dict) and data.get("@type") in ("Apartment", "SingleFamilyResidence", "Place"): price = data.get("offers", {}).get("price") if isinstance(data.get("offers"), dict) else None extracted.append({ "title": data.get("name"), "price": price, "address": data.get("address", {}).get("streetAddress", ""), "neighborhood": data.get("address", {}).get("addressLocality", ""), "source": "json-ld", "url": data.get("url") }) except Exception: continue # 2. Parse fallback HTML listing cards for card in soup.select(".listing-item"): title_el = card.select_one(".title") price_el = card.select_one(".price") addr_el = card.select_one(".address") link_el = card.select_one("a.link") if title_el and price_el: price_val = int("".join(c for c in price_el.get_text() if c.isdigit())) extracted.append({ "title": title_el.get_text().strip(), "price": price_val, "address": addr_el.get_text().strip() if addr_el else "", "neighborhood": "Astoria", "source": "dom-card", "url": link_el["href"] if link_el else "" }) return extracted if __name__ == "__main__": listings = extract_listings_from_html(SAMPLE_HTML_FIXTURE) print(f"Parsed {len(listings)} structured rental listings:") for idx, item in enumerate(listings, 1): print(f" {idx}. [{item['source'].upper()}] {item['title']} - ${item['price']}/mo ({item['url']})")

# models.py - Pydantic model for rental comps from pydantic import BaseModel, Field from typing import Optional class RentalProperty(BaseModel): property_id: str address: str zip_code: str rent: float bedrooms: int bathrooms: float sqft: Optional[int] = None url: str @property def price_per_sqft(self) -> Optional[float]: if self.sqft and self.sqft > 0: return round(self.rent / self.sqft, 2) return None prop = RentalProperty( property_id="prop-101", address="450 Elm St", zip_code="94103", rent=2800.0, bedrooms=2, bathrooms=1.0, sqft=850, url="https://example.com/prop-101" ) print("Model Validated:", prop.model_dump()) print("Price per sqft:", prop.price_per_sqft)

# comp_engine.py - Calculate market median comps & gross yield import statistics def calculate_neighborhood_comps(properties: list) -> dict: groups = {} for p in properties: key = (p["zip_code"], p["bedrooms"]) groups.setdefault(key, []).append(p["rent"]) comps = {} for (zip_code, beds), rents in groups.items(): comps[f"{zip_code}_{beds}bd"] = { "median_rent": round(statistics.median(rents), 2), "sample_size": len(rents) } return comps def calculate_gross_yield(monthly_rent: float, purchase_price: float) -> float: annual_income = monthly_rent * 12 return round((annual_income / purchase_price) * 100, 2) sample_comps = calculate_neighborhood_comps([ {"zip_code": "94103", "bedrooms": 2, "rent": 2900}, {"zip_code": "94103", "bedrooms": 2, "rent": 3100}, {"zip_code": "94103", "bedrooms": 2, "rent": 3000} ]) print("Calculated Comps:", sample_comps) print("Gross Yield ($3,000/mo on $450k purchase):", calculate_gross_yield(3000, 450000), "%")

# export_leads.py - Detect underpriced deals & export import pandas as pd def find_deals(properties: list, comps: dict, discount_threshold=0.12) -> list: deals = [] for p in properties: key = f"{p['zip_code']}_{p['bedrooms']}bd" if key in comps: median = comps[key]["median_rent"] discount = (median - p["rent"]) / median if discount >= discount_threshold: deals.append({ "address": p["address"], "rent": p["rent"], "median_comp": median, "discount_pct": round(discount * 100, 1), "url": p["url"] }) return deals deals = find_deals( [{"address": "123 Market St", "zip_code": "94103", "bedrooms": 2, "rent": 2400, "url": "https://sfbay.craigslist.org/sfc/apa/d/san-francisco-2bd-in-soma/7821001.html"}], {"94103_2bd": {"median_rent": 3000, "sample_size": 10}} ) df = pd.DataFrame(deals) df.to_csv("rental_leads.csv", index=False) print("Deals exported to rental_leads.csv:") print(df)

# init_db.py - SQLite rental schema import sqlite3 def init_rental_db(db_path="rental_history.db"): with sqlite3.connect(db_path) as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS properties ( property_id TEXT PRIMARY KEY, address TEXT, zip_code TEXT, bedrooms INTEGER, bathrooms REAL, sqft INTEGER ); """) conn.execute(""" CREATE TABLE IF NOT EXISTS rent_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, property_id TEXT, rent REAL, recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (property_id) REFERENCES properties (property_id) ); """) print("Database rental_history.db initialized with properties and rent_history tables.") init_rental_db()