# ============================================================ # FlipFinder AI — app.py (HF Spaces deployment) # Dan & Rotem · Final Project # ============================================================ # Pipeline: user filters → two-stage recommender (hard filter + # ideal-profile embedding ranking) → Gradient Boosting price # prediction → GenAI summary (Qwen 1.5B) # ============================================================ import spaces # required: this Space runs on ZeroGPU hardware import os import re import warnings warnings.filterwarnings("ignore") import numpy as np import pandas as pd import faiss import torch import gradio as gr from sentence_transformers import SentenceTransformer from transformers import pipeline, GenerationConfig from sklearn.ensemble import GradientBoostingRegressor # ============================================================ # BLOCK 1 — CONFIG # ============================================================ APP_VERSION = "v2-html-tables" # printed at startup so the deployed # build can be identified in the Logs tab print(f"🏠 FlipFinder starting - {APP_VERSION}") USE_HF_REPO = True # True on HF Spaces HF_DATASET_REPO = "rotemvahava/flipfinder-dataset" # dataset repo HF_DATASET_FILE = "flipfinder_final.csv" # the uploaded CSV LOCAL_CSV_PATH = "flipfinder_final.csv" # local fallback for testing EMBEDDINGS_PATH = "embeddings_bge.npy" # uploaded to the Space repo SEED = 42 # ============================================================ # BLOCK 2 — LOAD DATASET (HF dataset repo constraint) # ============================================================ print("⏳ Loading dataset ...") if USE_HF_REPO: from huggingface_hub import hf_hub_download csv_path = hf_hub_download( repo_id=HF_DATASET_REPO, filename=HF_DATASET_FILE, repo_type="dataset", ) df = pd.read_csv(csv_path) else: df = pd.read_csv(LOCAL_CSV_PATH) df = df.reset_index(drop=True) print(f"✅ Dataset loaded: {df.shape[0]:,} properties x {df.shape[1]} columns") # ============================================================ # BLOCK 3 — FEATURE ENGINEERING SAFETY NET # (labeled CSV already contains everything; recompute only if missing — # identical logic to the EDA notebook, vectorized for fast startup) # ============================================================ GRADE_GPA = {"A+": 4.3, "A": 4.0, "A-": 3.7, "B+": 3.3, "B": 3.0, "B-": 2.7} if "investment_label" not in df.columns: print("⏳ Engineered columns missing - recomputing (EDA logic) ...") df["zip3"] = df["zipcode"].astype(str).str[:3] # Comps: 3-level fallback medians (zipcode+bed+type → zipcode+bed → zip3+bed → zip3) g1 = df.groupby(["zipcode", "bedrooms", "property_type"])["price_per_sqft"] g2 = df.groupby(["zipcode", "bedrooms"])["price_per_sqft"] g3 = df.groupby(["zip3", "bedrooms"])["price_per_sqft"] g4 = df.groupby("zip3")["price_per_sqft"] med = np.where(g1.transform("count") >= 5, g1.transform("median"), np.where(g2.transform("count") >= 5, g2.transform("median"), np.where(g3.transform("count") >= 5, g3.transform("median"), g4.transform("median")))) df["local_median_ppsq"] = med df["arv"] = df["local_median_ppsq"] * df["sqft"] r1 = df.groupby(["zipcode", "bedrooms", "property_type"])["rent_estimate"] r2 = df.groupby(["zipcode", "bedrooms"])["rent_estimate"] r3 = df.groupby(["zip3", "bedrooms"])["rent_estimate"] r4 = df.groupby("zip3")["rent_estimate"] df["fair_rent"] = np.where(r1.transform("count") >= 5, r1.transform("median"), np.where(r2.transform("count") >= 5, r2.transform("median"), np.where(r3.transform("count") >= 5, r3.transform("median"), r4.transform("median")))) df["price_vs_market"] = (df["listed_price"] - df["arv"]) / df["arv"] df["gross_yield"] = (df["fair_rent"] * 12) / df["listed_price"] df["cap_rate"] = (df["fair_rent"] * 12 * 0.6) / df["listed_price"] df["price_to_rent_ratio"] = df["listed_price"] / (df["fair_rent"] * 12) df["property_age"] = 2024 - df["year_built"] df["ppsq_deviation"] = (df["price_per_sqft"] - df["local_median_ppsq"]) / df["local_median_ppsq"] * 100 for col, score in [("niche_overall_grade", "niche_overall_score"), ("school_rating", "school_score"), ("crime_safety_rating", "crime_safety_score"), ("housing_rating", "housing_score")]: df[score] = df[col].map(GRADE_GPA) VALID = list(GRADE_GPA.keys()) gidx = {g: i for i, g in enumerate(VALID)} # 0 = best (A+) def _label(row): g = row["niche_overall_grade"] if (row["listed_price"] > row["arv"] or row["gross_yield"] < 0.04 or row["days_on_market"] > 120 or g not in gidx or gidx[g] > gidx["B-"]): return "Bad Investment" if (row["price_vs_market"] <= -0.15 and row["year_built"] < 2020 and row["days_on_market"] <= 120 and row["property_type"] != "Condo"): return "Flip" if (row["price_vs_market"] <= -0.10 and row["gross_yield"] >= 0.08 and row["price_to_rent_ratio"] < 15 and row["property_type"] != "Condo" and g in gidx and gidx[g] <= gidx["B"]): return "BRRRR" good_hood = g in gidx and gidx[g] <= gidx["B"] strong = all(row[c] in gidx and gidx[row[c]] <= gidx["B+"] for c in ["school_rating", "housing_rating", "crime_safety_rating"]) if (0.05 <= row["gross_yield"] <= 0.08 and 20 <= row["days_on_market"] <= 120 and row["year_built"] >= 1990 and good_hood and strong): return "Buy and Hold" return "Uncategorized" df["investment_label"] = df.apply(_label, axis=1) print(f"✅ Investment labels: {df['investment_label'].value_counts().to_dict()}") # ============================================================ # BLOCK 4 — PROPERTY TEXT SERIALIZATION (identical to Part 3) # ============================================================ def property_to_text(row): return ( f"Investment profile: price vs market {row['price_vs_market']*100:.1f}%, " f"gross yield {row['gross_yield']*100:.1f}%, " f"cap rate {row['cap_rate']*100:.1f}%, " f"price to rent ratio {row['price_to_rent_ratio']:.1f}, " f"local price deviation {row['ppsq_deviation']:.1f}%. " f"Property: {int(row['bedrooms'])} bedrooms, {int(row['bathrooms'])} bathrooms, " f"{int(row['sqft']):,} sqft, {int(row['property_age'])} years old, " f"{int(row['days_on_market'])} days on market. " f"Neighborhood: overall score {row['niche_overall_score']:.1f}, " f"school {row['school_score']:.1f}, " f"crime safety {row['crime_safety_score']:.1f}, " f"housing {row['housing_score']:.1f}." ) # ============================================================ # BLOCK 5 — EMBEDDING MODEL (HF model repo constraint) + FAISS # ============================================================ EMBEDDING_MODEL_ID = "BAAI/bge-small-en-v1.5" # Part 3 winner print(f"⏳ Loading embedding model: {EMBEDDING_MODEL_ID} ...") embed_model = SentenceTransformer(EMBEDDING_MODEL_ID) print("✅ Embedding model ready") if os.path.exists(EMBEDDINGS_PATH): embeddings_bge = np.load(EMBEDDINGS_PATH).astype(np.float32) print(f"✅ Embeddings loaded: {embeddings_bge.shape}") else: print("⏳ Embeddings file not found - generating once ...") texts = df.apply(property_to_text, axis=1).tolist() embeddings_bge = embed_model.encode( texts, batch_size=64, show_progress_bar=True, convert_to_numpy=True ).astype(np.float32) np.save(EMBEDDINGS_PATH, embeddings_bge) print(f"✅ Embeddings generated: {embeddings_bge.shape}") # ============================================================ # BLOCK 6 — PRICE PREDICTION MODEL (Gradient Boosting, trained at startup) # ============================================================ print("⏳ Training price prediction model ...") df["log_listed_price"] = np.log1p(df["listed_price"]) df["log_sqft"] = np.log1p(df["sqft"]) PRICE_FEATURES = ["log_sqft", "bedrooms", "bathrooms", "lot_area_acres", "property_age", "days_on_market", "niche_overall_score", "school_score", "crime_safety_score", "housing_score", "fair_rent"] Xp = df[PRICE_FEATURES + ["property_type", "city"]].copy() Xp = pd.get_dummies(Xp, columns=["property_type"], drop_first=True) CITY_MEANS = df.groupby("city")["log_listed_price"].mean() GLOBAL_MEAN = df["log_listed_price"].mean() Xp["city"] = Xp["city"].map(CITY_MEANS).fillna(GLOBAL_MEAN) PRICE_COLUMNS = list(Xp.columns) # locked column order price_model = GradientBoostingRegressor( n_estimators=200, max_depth=4, learning_rate=0.1, random_state=SEED ) price_model.fit(Xp, df["log_listed_price"]) print("✅ Price model trained (Gradient Boosting)") def predict_price(prop_row): """Predict listing price (dollars) for one property row.""" x = {f: prop_row[f] for f in PRICE_FEATURES} x["city"] = CITY_MEANS.get(prop_row["city"], GLOBAL_MEAN) for col in PRICE_COLUMNS: if col.startswith("property_type_"): x[col] = 1 if col == f"property_type_{prop_row['property_type']}" else 0 xdf = pd.DataFrame([x])[PRICE_COLUMNS] return float(np.expm1(price_model.predict(xdf)[0])) # ============================================================ # BLOCK 7 — GENERATION MODEL (two-tier: HF Inference API + local fallback) # ============================================================ API_GEN_MODEL_ID = "Qwen/Qwen2.5-7B-Instruct" # served by the Inference API LOCAL_GEN_MODEL_ID = "Qwen/Qwen2.5-1.5B-Instruct" # small enough to run on this CPU # ── Generation strategy ─────────────────────────────────────── # Tier 1: the HF Inference API. The model executes on HuggingFace's servers, so a # summary takes ~1-2s regardless of this Space's (weak) CPU. The 7B variant # is used because small models like the 1.5B are not served by any provider. # Tier 2: the 1.5B model loaded locally on CPU. Slower (~15s) but always available, # so the app never breaks if the API is unreachable. HF_TOKEN = os.environ.get("HF_TOKEN") _inf_client = None if HF_TOKEN: try: from huggingface_hub import InferenceClient _inf_client = InferenceClient(token=HF_TOKEN) print(f"✅ Generation: HF Inference API ready ({API_GEN_MODEL_ID})") except Exception as e: print(f"⚠️ Inference API unavailable ({e}) - will use local CPU model") else: print("⚠️ HF_TOKEN not set - using local CPU model") print(f"⏳ Loading local fallback model: {LOCAL_GEN_MODEL_ID} ...") generator = pipeline( "text-generation", model=LOCAL_GEN_MODEL_ID, dtype=torch.float32, device="cpu", ) generator.tokenizer.clean_up_tokenization_spaces = False torch.set_num_threads(max(1, os.cpu_count() or 2)) GEN_CONFIG = GenerationConfig( do_sample=False, # greedy: faster on CPU and more disciplined max_new_tokens=110, repetition_penalty=1.05, pad_token_id=generator.tokenizer.eos_token_id, ) print("✅ Local fallback model ready (CPU)") GRADE_ORDER = {"A+": 12, "A": 11, "A-": 10, "B+": 9, "B": 8, "B-": 7, "C+": 6, "C": 5, "C-": 4, "D+": 3, "D": 2, "D-": 1} def _grade_rank(g): return GRADE_ORDER.get(str(g).strip(), 0) def _format_property(p): # Compact single-line format - fewer prompt tokens means faster inference pvm = p["price_vs_market"] pricing = (f"discount to market {abs(pvm):.1f}%" if pvm < 0 else f"premium to market {pvm:.1f}%") return ( f"Property {p['rank']}: {p['city']}, {p['state']} | {p['property_type']} | " f"${p['listed_price']:,} | {pricing} | " f"gross yield {p['gross_yield']}% | cap rate {p['cap_rate']}% | " f"neighborhood grade {p['niche_overall_grade']} | {p['days_on_market']} days on market" ) def _quick_facts(props): ranks = [p["rank"] for p in props] pvm = [p["price_vs_market"] for p in props] yields = [p["gross_yield"] for p in props] grades = [_grade_rank(p["niche_overall_grade"]) for p in props] return ("Pre-computed facts (treat as ground truth):\n" f" - Largest discount to market: Property {ranks[pvm.index(min(pvm))]}\n" f" - Highest gross yield: Property {ranks[yields.index(max(yields))]}\n" f" - Best neighborhood grade: Property {ranks[grades.index(max(grades))]}") def _build_messages(props): blocks = "\n".join(_format_property(p) for p in props) system = ( "You are FlipFinder, an expert real estate investment analysis engine. " "You write short, precise, factual property briefs for investors.\n" "Rules you must follow:\n" "- Neighborhood grades rank best to worst: A+ > A > A- > B+ > B > B- > C+ > C > C-.\n" "- 'Discount to market' is a percentage below comparable sales; larger discount = " "better flip upside.\n" "- 'Gross yield' and 'cap rate' are percentages; higher means better cash flow.\n" "- Only ever use these metric names: discount to market, gross yield, cap rate, " "neighborhood grade, days on market. Never invent other metrics.\n" "- You describe each property's strengths; you never advise buying or rank one " "property above another. No purchase recommendations.\n" "- Write flowing prose. Never use markdown, bold, bullets, headings, or labels." ) user = ( f"Here are three candidate properties.\n\n{blocks}\n\n{_quick_facts(props)}\n\n" "Write exactly three sentences of plain prose - one sentence per property, " "in order (Property 1, then 2, then 3).\n" "Each sentence names that property's one or two strongest points, using only " "the allowed metric names. Stay factual and neutral: describe strengths, but " "never tell the reader to buy, pick, or choose any property, and do not rank " "them against each other.\n\n" "Follow this style exactly:\n" "\"Property 1 stands out for its 22.4% discount to market, paired with a solid B+ " "neighborhood grade. Property 2 offers the strongest cash flow of the group, with a " "9.1% gross yield and a 5.5% cap rate. Property 3 combines an A- neighborhood grade " "with just 30 days on market, suggesting strong local demand.\"\n\n" "Now write your three sentences:" ) return [{"role": "system", "content": system}, {"role": "user", "content": user}] def _clean(text, max_sentences=3): # Strip lead-in filler ("Sure, here's...") text = re.sub(r"^\s*(sure[,!.]?|here('|)s[^:]*:?|certainly[,!.]?)\s*", "", text, flags=re.I) # Strip any "Sentence 1:" / "**Sentence 2:**" style labels the model may echo text = re.sub(r"\*{0,2}sentence\s*\d+\s*:?\*{0,2}\s*", "", text, flags=re.I) # Strip markdown emphasis and headings text = re.sub(r"[*_#`]+", "", text) text = re.sub(r"\s+", " ", text).strip().strip('"') sentences = re.split(r"(?<=[.!?])\s+", text) sentences = [s.strip() for s in sentences if s.strip()] # Drop a trailing sentence that was cut off mid-thought (no end punctuation) if sentences and not sentences[-1].endswith((".", "!", "?")): sentences = sentences[:-1] return " ".join(sentences[:max_sentences]).strip() # ZeroGPU hardware refuses to start a Space unless at least one @spaces.GPU # function is registered. This Space's ZeroGPU worker cannot actually attach a # device (torch.init fails with "No CUDA GPUs are available"), so real generation # runs on CPU. This stub exists solely to satisfy that startup check. @spaces.GPU(duration=1) def _zerogpu_startup_stub(): return "ok" def _generate_local(messages): """Fallback: run the model on this Space's CPU (~15s).""" out = generator(messages, generation_config=GEN_CONFIG, return_full_text=False) raw = out[0]["generated_text"] if isinstance(raw, list): raw = raw[-1]["content"] return raw def generate_investment_summary(top_3): """ Generate the 3-sentence analyst summary. Tries the HF Inference API first (~1-2s); falls back to the local CPU model if the API is unavailable, so the app always produces an analysis. """ messages = _build_messages(top_3) if _inf_client is not None: # Try the preferred model, then known-good alternatives if a provider # does not serve it. for model_id in (API_GEN_MODEL_ID, "meta-llama/Llama-3.1-8B-Instruct", "mistralai/Mistral-7B-Instruct-v0.3"): try: r = _inf_client.chat_completion( messages=messages, model=model_id, max_tokens=110, temperature=0.3, ) cleaned = _clean(r.choices[0].message.content) if cleaned: print(f"[gen] served by Inference API ({model_id})") return cleaned except Exception as e: print(f"[gen] API model {model_id} unavailable ({type(e).__name__}) - trying next") print("[gen] served by local CPU model") return _clean(_generate_local(messages)) # ============================================================ # BLOCK 8 — TWO-STAGE RECOMMENDER WITH FALLBACK (from Part 3 Block 11) # ============================================================ STRATEGY_MAP = { "Fix & Flip (High Discount)": "Flip", "Long-term Rental (High Cash Flow)": "Buy and Hold", "BRRRR Strategy": "BRRRR", } CITIES = set(df["city"].str.lower().unique()) STATES = set(df["state"].str.lower().unique()) def _apply_filters(base, state, city, max_price, property_type, label): """Hard filtering - Stage 1.""" d = base if state: d = d[d["state"].str.lower() == state.strip().lower()] if city: d = d[d["city"].str.lower() == city.strip().lower()] if property_type: d = d[d["property_type"] == property_type] if max_price: d = d[d["listed_price"] <= max_price] if label: d = d[d["investment_label"] == label] else: d = d[~d["investment_label"].isin(["Bad Investment", "Uncategorized"])] return d def build_ideal_property_text(pool): """The 'perfect deal' profile within the pool - Stage 2 query.""" return ( f"Investment profile: price vs market {pool['price_vs_market'].min()*100:.1f}%, " f"gross yield {pool['gross_yield'].max()*100:.1f}%, " f"cap rate {pool['cap_rate'].max()*100:.1f}%, " f"price to rent ratio {pool['price_to_rent_ratio'].min():.1f}, " f"local price deviation {pool['ppsq_deviation'].min():.1f}%. " f"Property: {int(pool['bedrooms'].median())} bedrooms, " f"{pool['bathrooms'].median():.1f} bathrooms, " f"{int(pool['sqft'].median()):,} sqft, " f"{int(pool['property_age'].median())} years old, 30 days on market. " f"Neighborhood: overall score 4.0, school 4.0, crime safety 4.0, housing 4.0." ) def recommend(state, city, max_price, property_type, strategy): """ Two-stage recommendation with graceful fallback. Returns (top_3: list[dict], notice: str). """ # Treat "Any" / empty as no filter if state and str(state).strip().lower() in ("any", ""): state = None if city and str(city).strip().lower() in ("any", ""): city = None if property_type and str(property_type).strip().lower() in ("any", ""): property_type = None if strategy and str(strategy).strip().lower() in ("any", ""): strategy = None if not max_price or float(max_price) <= 0: max_price = None label = STRATEGY_MAP.get(strategy) FALLBACK_SUFFIX = "here are the best 3 options that fit your description:" # Progressive relaxation - first combination with >= 3 results wins attempts = [ (dict(state=state, city=city, max_price=max_price, property_type=property_type, label=label), ""), (dict(state=state, city=city, max_price=(max_price * 1.5 if max_price else None), property_type=property_type, label=label), f"We could not find exact matches within your budget, so we widened it slightly - {FALLBACK_SUFFIX}"), (dict(state=state, city=city, max_price=None, property_type=property_type, label=label), f"We could not find exact matches within your budget in this location - {FALLBACK_SUFFIX}"), (dict(state=state, city=city, max_price=max_price, property_type=None, label=label), f"We could not find exact matches for that property type - {FALLBACK_SUFFIX}"), (dict(state=state, city=None, max_price=max_price, property_type=property_type, label=label), f"We could not find exact matches in that city, so we searched across the state - {FALLBACK_SUFFIX}"), (dict(state=None, city=None, max_price=max_price, property_type=property_type, label=label), f"We could not find exact matches in that location, so we searched other markets - {FALLBACK_SUFFIX}"), (dict(state=None, city=None, max_price=None, property_type=None, label=label), f"We could not find exact matches for your filters - {FALLBACK_SUFFIX}"), (dict(state=None, city=None, max_price=None, property_type=None, label=None), f"We could not find exact matches - {FALLBACK_SUFFIX}"), ] pool, notice = None, "" for filt, msg in attempts: cand = _apply_filters(df, **filt) if len(cand) >= 3: pool, notice = cand, msg break if pool is None: return [], "❌ No properties available." # Stage 2 - mini FAISS on filtered pool, ranked vs ideal profile idxs = pool.index.tolist() emb = embeddings_bge[idxs].copy() faiss.normalize_L2(emb) mini = faiss.IndexFlatIP(emb.shape[1]) mini.add(emb) q = embed_model.encode([build_ideal_property_text(pool)], convert_to_numpy=True).astype(np.float32) faiss.normalize_L2(q) scores, top = mini.search(q, 3) results = [] for rank, (score, i) in enumerate(zip(scores[0], top[0]), start=1): p = pool.iloc[i] results.append({ "rank": rank, "street": p["street"], "city": p["city"], "state": p["state"], "zipcode": str(p["zipcode"]), "property_type": p["property_type"], "investment_label": p["investment_label"], "listed_price": int(p["listed_price"]), "sqft": int(p["sqft"]), "bedrooms": int(p["bedrooms"]), "bathrooms": float(p["bathrooms"]), "year_built": int(p["year_built"]), "gross_yield": round(float(p["gross_yield"]) * 100, 1), "cap_rate": round(float(p["cap_rate"]) * 100, 1), "price_vs_market": round(float(p["price_vs_market"]) * 100, 1), "niche_overall_grade": p["niche_overall_grade"], "days_on_market": int(p["days_on_market"]), "similarity_score": round(float(score), 4), "_row": p, # for price prediction }) return results, notice # ============================================================ # BLOCK 9 — GRADIO WRAPPER (progressive output: table → prices → summary) # ============================================================ TABLE_COLS = ["rank", "street", "city", "state", "sqft", "property_type", "investment_label", "listed_price", "gross_yield", "cap_rate", "niche_overall_grade", "days_on_market"] def _parse_budget(budget): """Accept 'Any', '300,000', '$300000', 300000 -> float or None.""" if budget is None: return None if isinstance(budget, (int, float)): return float(budget) if budget > 0 else None s = str(budget).strip().lower().replace(",", "").replace("$", "") if s in ("", "any"): return None try: v = float(s) # reject nan, inf and absurd magnitudes if v != v or v in (float("inf"), float("-inf")) or v > 1e9: return None return v if v > 0 else None except ValueError: return None def _is_set(v): return v not in (None, "") and str(v).strip().lower() != "any" def _df_to_html(df, title): """ Render a DataFrame as a self-contained horizontally-scrollable HTML table. Gradio's dataframe component cannot be reliably made to scroll sideways on narrow screens, so the results tables are emitted as HTML with the scroll container and styling defined inline - inline styles are not overridden by Gradio's own stylesheet. """ if df is None or len(df) == 0: return "" head = "".join( f'{c}' for c in df.columns ) body = "" for _, row in df.iterrows(): cells = "".join( f'{v}' for v in row ) body += f"{cells}" return f"""
{title}
{head}{body}
Swipe sideways to see all columns
""" def run_flipfinder(state, city, budget, property_type, strategy): budget = _parse_budget(budget) # No filters set is a valid query: it means "show me the best deals anywhere". # The recommender already excludes Bad Investment / Uncategorized in that case. top_3, notice = recommend(state, city, budget, property_type, strategy) if not top_3: yield notice or "No matching properties found.", "", "", "" return # --- Properties table --- table = pd.DataFrame([{k: p[k] for k in TABLE_COLS} for p in top_3]) table["listed_price"] = table["listed_price"].apply(lambda x: f"${x:,}") table["sqft"] = table["sqft"].apply(lambda x: f"{x:,}") table["gross_yield"] = table["gross_yield"].apply(lambda x: f"{x:.1f}%") table["cap_rate"] = table["cap_rate"].apply(lambda x: f"{x:.1f}%") table = table.rename(columns={ "rank": "#", "street": "Address", "city": "City", "state": "State", "sqft": "Sqft", "property_type": "Type", "investment_label": "Strategy", "listed_price": "Listed Price", "gross_yield": "Gross Yield", "cap_rate": "Cap Rate", "niche_overall_grade": "Neighborhood", "days_on_market": "Days Listed", }) # --- Price predictions --- rows = [] for p in top_3: pred = predict_price(p["_row"]) gap = (p["listed_price"] - pred) / pred * 100 # Negative gap = listed BELOW the model's predicted value = a bargain if gap <= -3: verdict = f"🟢 UNDERVALUED by {abs(gap):.1f}%" elif gap >= 3: verdict = f"🔴 OVERVALUED by {gap:.1f}%" else: verdict = f"⚪ Fairly priced ({gap:+.1f}%)" # price_vs_market: negative = listed below comparable sales (the discount # the AI analysis refers to) pvm = p["price_vs_market"] vs_comps = (f"🟢 {abs(pvm):.1f}% below" if pvm < 0 else f"🔴 {pvm:.1f}% above") rows.append({ "Property": f"#{p['rank']} — {p['street']}, {p['city']}", "Listed Price": f"${p['listed_price']:,}", "Vs Comparable Sales": vs_comps, "AI Predicted Value": f"${pred:,.0f}", "Verdict": verdict, }) price_table = pd.DataFrame(rows) notice_md = f"**{notice}**" if notice else "" # Yield 1 - properties + price predictions appear immediately props_html = _df_to_html(table, "🏆 Top 3 Recommended Properties") prices_html = _df_to_html(price_table, "💰 AI Price Prediction (Gradient Boosting)") yield notice_md, props_html, prices_html, "⏳ Writing your investment analysis..." # --- GenAI summary (HF Inference API: ~1-2s) --- try: summary = generate_investment_summary(top_3) except Exception as e: summary = f"AI summary unavailable: {e}" yield notice_md, props_html, prices_html, summary # ============================================================ # BLOCK 9b — INVESTMENT CALCULATOR (pure math, standalone) # ============================================================ def calculate_investment( purchase_price, monthly_rent, rehab_budget=0, down_payment_pct=25, interest_rate=7.0, loan_term_years=30, property_tax_yr=None, insurance_yr=1200, maintenance_pct=5, vacancy_pct=5, management_pct=0, closing_costs_pct=3, ): """ Underwrite a single rental/flip deal. Only purchase_price and monthly_rent are required; every other input has a sensible default so the calculator always produces a complete analysis. """ try: price = float(purchase_price) rent = float(monthly_rent) except (TypeError, ValueError): return "⚠️ Please enter a valid purchase price and monthly rent." if price <= 0 or rent <= 0: return "⚠️ Purchase price and monthly rent must be greater than zero." if price != price or rent != rent or price in (float("inf"), float("-inf")) or rent in (float("inf"), float("-inf")): return "⚠️ Please enter a valid purchase price and monthly rent." if price < 1000: return "⚠️ Please enter a realistic purchase price (at least $1,000)." if price > 100_000_000 or rent > 1_000_000: return "⚠️ Those figures look unrealistic - please check the purchase price and monthly rent." rehab = max(0.0, float(rehab_budget or 0)) # Clamp assumptions to sane ranges so out-of-range entries cannot produce # nonsense (negative loans, absurd returns, infinite values). Every clamp is # reported back to the user rather than applied silently. notes = [] def _clamp(value, default, lo, hi, label, unit="%"): if value is None or value == "": return float(default) v = float(value) c = min(max(v, lo), hi) if c != v: notes.append(f"{label} of {v:g}{unit} is outside the valid range " f"({lo:g}{unit}–{hi:g}{unit}) - adjusted to {c:g}{unit}.") return c dp_pct = _clamp(down_payment_pct, 25, 0, 100, "Down payment") rate = _clamp(interest_rate, 7.0, 0, 30, "Interest rate") term = int(_clamp(loan_term_years, 30, 1, 50, "Loan term", " years")) maint_pct = _clamp(maintenance_pct, 5, 0, 100, "Maintenance") vac_pct = _clamp(vacancy_pct, 5, 0, 100, "Vacancy") mgmt_pct = _clamp(management_pct, 0, 0, 100, "Management") closing_pct = _clamp(closing_costs_pct, 3, 0, 20, "Closing costs") tax_yr = float(property_tax_yr) if property_tax_yr not in (None, "") else price * 0.011 if tax_yr < 0: notes.append("Property tax cannot be negative - treated as 0.") tax_yr = 0.0 ins_yr = float(insurance_yr if insurance_yr is not None else 1200) if ins_yr < 0: notes.append("Insurance cannot be negative - treated as 0.") ins_yr = 0.0 if rehab_budget is not None and float(rehab_budget or 0) < 0: notes.append("Rehab budget cannot be negative - treated as 0.") down_payment = price * dp_pct / 100 loan_amount = price - down_payment closing_costs = price * closing_pct / 100 monthly_rate = rate / 100 / 12 n_payments = term * 12 if loan_amount <= 0: mortgage = 0.0 elif monthly_rate == 0: mortgage = loan_amount / n_payments else: mortgage = loan_amount * (monthly_rate * (1 + monthly_rate) ** n_payments) \ / ((1 + monthly_rate) ** n_payments - 1) tax_m = tax_yr / 12 ins_m = ins_yr / 12 maint_m = rent * maint_pct / 100 vac_m = rent * vac_pct / 100 mgmt_m = rent * mgmt_pct / 100 operating_expenses = tax_m + ins_m + maint_m + vac_m + mgmt_m noi_month = rent - operating_expenses # excludes mortgage monthly_cash_flow = noi_month - mortgage annual_cash_flow = monthly_cash_flow * 12 annual_noi = noi_month * 12 total_cash_invested = down_payment + closing_costs + rehab all_in_cost = price + rehab cap_rate = annual_noi / all_in_cost * 100 if all_in_cost else 0 gross_yield = rent * 12 / all_in_cost * 100 if all_in_cost else 0 coc_return = annual_cash_flow / total_cash_invested * 100 if total_cash_invested else 0 dscr = noi_month / mortgage if mortgage > 0 else float("inf") breakeven_rent = mortgage + operating_expenses cf_flag = "🟢" if monthly_cash_flow >= 0 else "🔴" dscr_str = "n/a (no loan)" if dscr == float("inf") else f"{dscr:.2f}" # Plausibility checks - combinations that are arithmetically valid but do not # reflect a realistic deal. These are advisory, not errors. warnings = [] annual_rent = rent * 12 if annual_rent / price > 0.35: warnings.append(f"A yearly rent of ${annual_rent:,.0f} on a ${price:,.0f} property " f"({gross_yield:.0f}% gross yield) is far above typical market levels - " "double-check the rent figure.") if annual_rent / price < 0.02: warnings.append(f"A gross yield of {gross_yield:.1f}% is very low for a rental - " "double-check the rent figure.") if maint_pct + vac_pct + mgmt_pct >= 60: warnings.append(f"Maintenance, vacancy and management together consume " f"{maint_pct + vac_pct + mgmt_pct:.0f}% of rent, which is unusually high.") if rehab > price: warnings.append(f"The rehab budget (${rehab:,.0f}) exceeds the purchase price - " "make sure that is intended.") if dp_pct == 0: warnings.append("A 0% down payment means financing the full purchase price; " "most investment lenders require 20-25%.") if term <= 5 and dp_pct < 100: warnings.append(f"A {term}-year term makes payments very high; " "investment mortgages are typically 15-30 years.") if monthly_cash_flow < 0: warnings.append("This deal is cash-flow negative: the rent does not cover the " "mortgage and running costs.") if dscr != float("inf") and dscr < 1.0: warnings.append(f"A DSCR of {dscr:.2f} is below 1.0 - most lenders require at " "least 1.20 to approve an investment loan.") notes_block = "" if notes: notes_block += "\n> **Adjusted inputs**\n" + "".join(f">\n> - {n}\n" for n in notes) if warnings: notes_block += "\n> **Worth checking**\n" + "".join(f">\n> - {w}\n" for w in warnings) return notes_block + f"""## 📊 Investment Analysis **Monthly cash flow:** {cf_flag} ${monthly_cash_flow:,.0f}    (${annual_cash_flow:,.0f}/year) | Return metric | Value | |---|---| | Cash-on-cash return | **{coc_return:.1f}%** | | Cap rate | {cap_rate:.1f}% | | Gross yield | {gross_yield:.1f}% | | DSCR (debt coverage) | {dscr_str} | | Cost breakdown | Amount | |---|---| | Down payment ({dp_pct:.0f}%) | ${down_payment:,.0f} | | Closing costs ({closing_pct:.0f}%) | ${closing_costs:,.0f} | | Rehab budget | ${rehab:,.0f} | | **Total cash invested** | **${total_cash_invested:,.0f}** | | All-in cost (price + rehab) | ${all_in_cost:,.0f} | | Monthly detail | Amount | |---|---| | Rent | ${rent:,.0f} | | Mortgage payment | ${mortgage:,.0f} | | Operating expenses | ${operating_expenses:,.0f} | | Break-even rent | ${breakeven_rent:,.0f} | *Operating expenses = property tax + insurance + maintenance + vacancy + management. Defaults are used for any field left blank. This is an estimate, not financial advice.* """ # ============================================================ # BLOCK 10 — GRADIO UI (dark professional theme) # ============================================================ CUSTOM_CSS = """ @import url('https://fonts.googleapis.com/css2?family=Inter:wght@400;500;600;700;800&display=swap'); /* ── Base ─────────────────────────────────────────── */ .gradio-container { background: #000000 !important; color: #e8eaed !important; font-family: 'Inter', -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif !important; max-width: 1240px !important; margin: 0 auto !important; } * { font-family: 'Inter', -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif !important; } /* ── Typography hierarchy ─────────────────────────── */ #app-title { font-size: 2.6rem !important; font-weight: 800 !important; letter-spacing: -0.03em !important; color: #ffffff !important; margin-bottom: 2px !important; } #header-sub { color: #9aa0a6 !important; font-size: 1.05rem !important; font-weight: 400 !important; } .prose h3, h3 { color: #ffffff !important; font-size: 1.35rem !important; font-weight: 700 !important; letter-spacing: -0.02em !important; margin-top: 10px !important; } label span, .gr-block label, .block label span { color: #c9cdd3 !important; font-size: 0.95rem !important; font-weight: 600 !important; letter-spacing: 0.01em !important; } /* ── Robinhood green primary button ───────────────── */ .gr-button-primary, button.primary { background: #00c805 !important; border: none !important; color: #000000 !important; font-weight: 700 !important; font-size: 1.05rem !important; letter-spacing: 0.01em !important; border-radius: 28px !important; padding: 12px 28px !important; transition: all .15s ease !important; } button.primary:hover { background: #00e206 !important; box-shadow: 0 0 22px rgba(0, 200, 5, 0.35) !important; transform: translateY(-1px) !important; } /* ── Panels, inputs ───────────────────────────────── */ .gr-box, .gr-panel, .gr-form, .gr-input, .block, textarea, input, select { background: #0b0d0f !important; color: #e8eaed !important; border: 1px solid #1c2126 !important; border-radius: 12px !important; } input:focus, select:focus, textarea:focus { border-color: #00c805 !important; box-shadow: 0 0 0 2px rgba(0, 200, 5, 0.18) !important; } /* ── Tables ───────────────────────────────────────── */ table { background: #0b0d0f !important; color: #e8eaed !important; border-collapse: collapse !important; font-size: 0.94rem !important; } thead, thead th { background: #000000 !important; color: #9aa0a6 !important; font-weight: 600 !important; text-transform: uppercase !important; font-size: 0.78rem !important; letter-spacing: 0.06em !important; border-bottom: 1px solid #1c2126 !important; } tbody td { border-bottom: 1px solid #14181c !important; padding: 12px 10px !important; } tbody tr:hover { background: #101418 !important; } /* ── Accordion (glossary) ─────────────────────────── */ .gr-accordion, .accordion { background: #0b0d0f !important; border: 1px solid #1c2126 !important; border-radius: 12px !important; } /* ── Quick starters ───────────────────────────────── */ .gr-examples, .examples { background: transparent !important; } /* ── Mobile / narrow screens ──────────────────────── */ @media (max-width: 820px) { /* Stop the page itself from ever being wider than the screen */ html, body, .gradio-container, gradio-app { max-width: 100vw !important; width: 100% !important; overflow-x: hidden !important; margin: 0 !important; } .gradio-container { padding: 0 6px !important; } #app-title { font-size: 1.8rem !important; } #header-sub { font-size: 0.88rem !important; } /* Nothing inside may force the page wider - except tables, which are allowed to stay wide and scroll inside their own wrapper (see below). */ .gradio-container *:not(table):not(thead):not(tbody):not(tr):not(th):not(td) { max-width: 100% !important; } /* Stack rows vertically instead of squeezing side by side */ .gr-row, .row, div[class*="svelte"][class*="row"] { flex-direction: column !important; flex-wrap: wrap !important; gap: 8px !important; } .gr-row > *, .row > * { min-width: 0 !important; width: 100% !important; flex: 1 1 100% !important; } /* Results tables are rendered as HTML with inline styles (see _df_to_html) so they scroll reliably on narrow screens without fighting Gradio's CSS. */ .gr-button-primary, button.primary { width: 100% !important; font-size: 1rem !important; padding: 14px 18px !important; } .prose h3, h3 { font-size: 1.1rem !important; } /* Textboxes and number inputs full width, readable font */ input, select, textarea { font-size: 16px !important; } /* 16px stops iOS zoom */ } footer { display: none !important; } """ with gr.Blocks(css=CUSTOM_CSS, title="FlipFinder AI", theme=gr.themes.Base(primary_hue="green", neutral_hue="zinc")) as demo: # Ensure a proper mobile viewport (Gradio does not always set one, which # makes the page render at desktop width and appear "cut in half" on phones). gr.HTML( """ """ ) gr.HTML( """

FlipFinder

AI-Powered Real Estate Investment Advisor

""" ) with gr.Tabs(): with gr.Tab("🔍 Find Properties"): # State -> cities mapping for cascading dropdowns STATE_CITIES = { s: sorted(df.loc[df["state"] == s, "city"].unique().tolist()) for s in sorted(df["state"].unique().tolist()) } CITY_STATE = {c: s for s, cities in STATE_CITIES.items() for c in cities} ALL_STATES = ["Any"] + sorted(STATE_CITIES.keys()) ALL_CITIES = ["Any"] + sorted(CITY_STATE.keys()) with gr.Row(): state_input = gr.Dropdown( choices=ALL_STATES, value="Any", label="🗺️ State", ) city_input = gr.Dropdown( choices=ALL_CITIES, value="Any", label="📍 City", ) budget_input = gr.Dropdown( choices=["Any", "150,000", "250,000", "300,000", "500,000", "750,000", "1,000,000", "1,500,000"], value="Any", label="💰 Max Budget ($)", allow_custom_value=True, ) def _on_state_change(state, city): """Choosing a state narrows the city list to that state's cities.""" if not state or state == "Any": return gr.update(choices=ALL_CITIES) allowed = ["Any"] + STATE_CITIES[state] # keep the current city if it belongs to this state, otherwise reset new_value = city if city in allowed else "Any" return gr.update(choices=allowed, value=new_value) def _on_city_change(city, state): """Choosing a city locks the state to the city's state.""" if not city or city == "Any": return gr.update() return gr.update(value=CITY_STATE.get(city, state)) state_input.change(_on_state_change, inputs=[state_input, city_input], outputs=city_input) city_input.change(_on_city_change, inputs=[city_input, state_input], outputs=state_input) with gr.Row(): property_type_input = gr.Dropdown( choices=["Any"] + sorted(df["property_type"].unique().tolist()), value="Any", label="🏘️ Property Type", ) strategy_input = gr.Dropdown( choices=["Any"] + list(STRATEGY_MAP.keys()), value="Any", label="📊 Investment Strategy", ) submit_btn = gr.Button("🔍 Find Properties", variant="primary", size="lg") gr.Examples( label="✨ Quick Starters — click to auto-fill, then press Find Properties", examples=[ ["Texas", "Houston", "300,000", "Single Family", "Fix & Flip (High Discount)"], ["Florida", "Miami", "500,000", "Condo", "Long-term Rental (High Cash Flow)"], ["Ohio", "Cleveland", "250,000", "Single Family", "BRRRR Strategy"], ], inputs=[state_input, city_input, budget_input, property_type_input, strategy_input], cache_examples=False, ) gr.Markdown("### Results") notice_md = gr.Markdown("") # Rendered as HTML rather than gr.Dataframe so we control the # scroll container directly - Gradio's own dataframe styling # cannot be reliably overridden on narrow screens. properties_table = gr.HTML("") price_table = gr.HTML("") summary_box = gr.Textbox( label="🤖 AI Investment Analysis", lines=4, interactive=False, ) with gr.Accordion("📖 What do these terms mean?", open=False): gr.Markdown( """ **Sqft** — the interior living area of the property in square feet. **Vs Comparable Sales** — how the asking price compares to what similar homes (same zip code, same bedroom count, same type) actually sell for per square foot. 🟢 *below* means it is listed cheaper than its comparables — this is the "discount" the AI analysis refers to. 🔴 *above* means you would be paying a premium. **Gross Yield** — one year of rent as a percentage of the purchase price. A $200,000 home renting for $1,500/month yields 9%. Higher is better cash flow. **Cap Rate** — gross yield after subtracting roughly 40% for running costs (tax, insurance, repairs, vacancy). This is the more realistic return figure. **Neighborhood** — an overall area grade from A+ (best) down to B- (weakest we list), combining school quality, safety, and housing. **Days Listed** — how long the property has been on the market. A long time on market can signal a problem, or an owner willing to negotiate. **AI Predicted Value** — what our price model thinks the home is genuinely worth, based on its size, age, location and neighborhood — *ignoring* the asking price. If the asking price sits well below this, the property is flagged 🟢 **UNDERVALUED**; well above it, 🔴 **OVERVALUED**. --- **The three strategies** **Fix & Flip** — buy clearly under market value, renovate, sell on. The number that matters is the discount to comparable sales. **Long-term Rental (Buy and Hold)** — keep the property and rent it out for years. The numbers that matter are gross yield, cap rate, and a solid neighborhood. **BRRRR** — Buy, Rehab, Rent, Refinance, Repeat. Needs *both* a discount *and* strong rent, so you can refinance your money back out and buy the next one. """ ) submit_btn.click( fn=run_flipfinder, inputs=[state_input, city_input, budget_input, property_type_input, strategy_input], outputs=[notice_md, properties_table, price_table, summary_box], ) with gr.Tab("🧮 Investment Calculator"): gr.Markdown( "### Underwrite any deal\n" "Enter a purchase price and expected monthly rent to get a full " "cash-flow and returns analysis. Every other field is optional — " "sensible defaults are used if you leave it blank." ) with gr.Row(): calc_price = gr.Number(label="Purchase Price ($) *", value=300000, precision=0, minimum=1000, maximum=100_000_000) calc_rent = gr.Number(label="Monthly Rent ($) *", value=2200, precision=0, minimum=1, maximum=1_000_000) calc_rehab = gr.Number(label="Rehab Budget ($)", value=0, precision=0, minimum=0, maximum=10_000_000) with gr.Accordion("Financing & expense assumptions (optional)", open=True): with gr.Row(): calc_dp = gr.Number(label="Down Payment (%)", value=25, minimum=0, maximum=100) calc_rate = gr.Number(label="Interest Rate (%)", value=7.0, minimum=0, maximum=30) calc_term = gr.Number(label="Loan Term (years)", value=30, precision=0, minimum=1, maximum=50) with gr.Row(): calc_tax = gr.Number(label="Property Tax / year ($)", value=None, minimum=0) calc_ins = gr.Number(label="Insurance / year ($)", value=1200, precision=0, minimum=0) calc_close = gr.Number(label="Closing Costs (%)", value=3, minimum=0, maximum=20) with gr.Row(): calc_maint = gr.Number(label="Maintenance (% of rent)", value=5, minimum=0, maximum=100) calc_vac = gr.Number(label="Vacancy (% of rent)", value=5, minimum=0, maximum=100) calc_mgmt = gr.Number(label="Management (% of rent)", value=0, minimum=0, maximum=100) calc_btn = gr.Button("Calculate", variant="primary", size="lg") calc_output = gr.Markdown("") calc_btn.click( fn=calculate_investment, inputs=[calc_price, calc_rent, calc_rehab, calc_dp, calc_rate, calc_term, calc_tax, calc_ins, calc_maint, calc_vac, calc_mgmt, calc_close], outputs=calc_output, ) if __name__ == "__main__": demo.launch()