topaLE
/
ghostfolio
mirror of https://github.com/ghostfolio/ghostfolio
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
4599 Commits
19 Branches
688 Tags
297 MiB
 
 
 
 
 
Tree: f992f7a86f
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'f992f7a86f'
${ noResults }
ghostfolio/agent/tools/wealth_bridge.py

530 lines
20 KiB
Raw Blame History

"""
Wealth Bridge Tool — AgentForge integration
============================================
Bridges live Ghostfolio portfolio data with real estate purchasing power.
Three capabilities:
1. calculate_down_payment_power(portfolio_value, target_cities)
— which markets can your portfolio fund a 20% down payment?
2. calculate_job_offer_affordability(offer_salary, offer_city,
current_salary, current_city)
— is the job offer a real raise in purchasing power terms?
3. get_portfolio_real_estate_summary(target_cities)
— master function: reads live portfolio then runs down payment calc
Data routing:
Austin TX areas → _MOCK_SNAPSHOTS in real_estate.py (real ACTRIS data)
All other cities → teleport_api.py (live Teleport API + fallback)
Mortgage assumption: 30-year fixed at 6.95%, 20% down, payment × 1.25
for estimated taxes + insurance.
"""
import asyncio
from typing import Optional
from agent.tools.real_estate import _MOCK_SNAPSHOTS, _normalize_city
from agent.tools.teleport_api import (
HARDCODED_FALLBACK,
_is_austin_area,
get_city_housing_data,
)
# ---------------------------------------------------------------------------
# COL index values for Austin TX sub-markets (ACTRIS coverage areas)
# National average = 100; lower = more affordable
# ---------------------------------------------------------------------------
_AUSTIN_COL_INDEX: dict[str, float] = {
"austin": 95.4,
"travis_county": 95.4,
"austin_msa": 91.0,
"williamson_county": 88.2,
"hays_county": 82.1,
"bastrop_county": 78.3,
"caldwell_county": 71.2,
}
# State income tax lookup (no tax = not in this dict)
_STATE_INCOME_TAX: dict[str, float] = {
"CA": 0.093, "NY": 0.0685, "OR": 0.099, "MN": 0.0985,
"NJ": 0.1075, "VT": 0.0875, "DC": 0.0895, "HI": 0.11,
"ME": 0.0715, "IA": 0.0575, "SC": 0.07, "CT": 0.0699,
"WI": 0.0765, "MA": 0.05, "IL": 0.0495, "IN": 0.032,
"MI": 0.0425, "GA": 0.055, "NC": 0.0499, "VA": 0.0575,
"MD": 0.0575, "CO": 0.044, "AZ": 0.025, "UT": 0.0485,
"KS": 0.057, "MO": 0.0495, "OH": 0.0399, "PA": 0.0307,
"NM": 0.059, "LA": 0.06, "MS": 0.05, "AL": 0.05,
"AR": 0.044, "NE": 0.0664, "ID": 0.058, "MT": 0.069,
"ND": 0.029, "OK": 0.0475, "KY": 0.045,
}
_NO_INCOME_TAX_STATES = {"TX", "WA", "FL", "NV", "WY", "SD", "AK", "NH", "TN"}
# Default target markets (all 7 Austin ACTRIS areas)
_DEFAULT_AUSTIN_MARKETS = [
"austin", "travis_county", "williamson_county",
"hays_county", "bastrop_county", "caldwell_county", "austin_msa",
]
# Mortgage constants
_MORTGAGE_RATE = 0.0695
_MORTGAGE_TERM = 360 # 30 years in months
_TAX_INSURANCE_MULTIPLIER = 1.25
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _monthly_payment(price: float) -> float:
"""Calculates estimated total monthly payment (PITI) at 6.95% 30yr, 20% down."""
principal = price * 0.80
r = _MORTGAGE_RATE / 12
n = _MORTGAGE_TERM
base_payment = principal * (r * (1 + r) ** n) / ((1 + r) ** n - 1)
return round(base_payment * _TAX_INSURANCE_MULTIPLIER, 0)
def _get_austin_data(city_key: str) -> Optional[dict]:
"""Reads ACTRIS snapshot for a given city_key from real_estate._MOCK_SNAPSHOTS."""
return _MOCK_SNAPSHOTS.get(city_key)
def _resolve_city_data_sync(city_name: str) -> tuple[str, dict]:
"""
Synchronously resolves city data for wealth bridge calculations.
Returns (display_name, data_dict).
Austin areas: uses _MOCK_SNAPSHOTS (real ACTRIS data, sync)
Other cities: uses HARDCODED_FALLBACK (sync) or generic estimate
"""
if _is_austin_area(city_name):
city_key = _normalize_city(city_name)
snap = _MOCK_SNAPSHOTS.get(city_key, {})
display = snap.get("city", city_name)
return display, snap
# Non-Austin: look up in fallback dict
lower = city_name.lower().strip()
slug_guess = lower.replace(" ", "-")
# Direct slug match
if slug_guess in HARDCODED_FALLBACK:
return HARDCODED_FALLBACK[slug_guess]["city"], dict(HARDCODED_FALLBACK[slug_guess])
# Partial match
for slug, data in HARDCODED_FALLBACK.items():
if lower in data["city"].lower() or slug.replace("-", " ") in lower:
return data["city"], dict(data)
# Generic
return city_name, {
"city": city_name,
"median_price": 500_000,
"MedianRentMonthly": 2000,
"col_index": 100.0,
"AffordabilityScore": 5.0,
"data_source": "Estimate (city not in database)",
}
def _col_index_for_city(city_name: str, city_data: dict) -> float:
"""
Returns the COL index for a city (100 = US average).
Austin areas: use _AUSTIN_COL_INDEX table.
Others: use col_index from city_data, or derive from Teleport col_score.
"""
if _is_austin_area(city_name):
city_key = _normalize_city(city_name)
return _AUSTIN_COL_INDEX.get(city_key, 95.4)
if "col_index" in city_data:
return float(city_data["col_index"])
# Derive from Teleport col_score (0–10): 10=cheap, 0=expensive
col_score = city_data.get("col_score", 5.0)
return round((10.0 - col_score) * 18.0 + 20.0, 1)
def _state_tax_note(city_a: str, city_b: str) -> str:
"""Builds a human-readable state income tax note for two cities."""
def _state_code(city: str) -> Optional[str]:
# Simple heuristics from known cities
city_lower = city.lower()
state_map = {
"tx": "TX", "austin": "TX", "dallas": "TX", "houston": "TX", "san antonio": "TX",
"wa": "WA", "seattle": "WA",
"fl": "FL", "miami": "FL", "orlando": "FL", "tampa": "FL",
"nv": "NV", "las vegas": "NV",
"ca": "CA", "san francisco": "CA", "los angeles": "CA", "san diego": "CA",
"ny": "NY", "new york": "NY", "brooklyn": "NY",
"co": "CO", "denver": "CO",
"il": "IL", "chicago": "IL",
"ma": "MA", "boston": "MA",
"tn": "TN", "nashville": "TN",
"ga": "GA", "atlanta": "GA",
"or": "OR", "portland": "OR",
"az": "AZ", "phoenix": "AZ",
"mn": "MN", "minneapolis": "MN",
"nc": "NC", "charlotte": "NC",
"va": "VA", "arlington": "VA",
}
for keyword, code in state_map.items():
if keyword in city_lower:
return code
return None
state_a = _state_code(city_a)
state_b = _state_code(city_b)
if not state_a or not state_b:
return "State income tax comparison not available for one or both cities."
a_notax = state_a in _NO_INCOME_TAX_STATES
b_notax = state_b in _NO_INCOME_TAX_STATES
if a_notax and b_notax:
return (
f"Both {state_a} and {state_b} have no state income tax, "
"so this does not affect the comparison."
)
if a_notax and not b_notax:
rate = _STATE_INCOME_TAX.get(state_b, 0.05) * 100
return (
f"{state_a} has no state income tax. {state_b} charges ~{rate:.1f}% — "
"this makes the offer worth even less than the purchasing power calculation shows."
)
if not a_notax and b_notax:
rate = _STATE_INCOME_TAX.get(state_a, 0.05) * 100
return (
f"{state_b} has no state income tax vs {state_a}'s ~{rate:.1f}% — "
"the move actually improves your after-tax position beyond the COL calculation."
)
rate_a = _STATE_INCOME_TAX.get(state_a, 0.05) * 100
rate_b = _STATE_INCOME_TAX.get(state_b, 0.05) * 100
return (
f"{state_a} taxes income at ~{rate_a:.1f}%, {state_b} at ~{rate_b:.1f}%. "
"Factor this into your take-home pay comparison."
)
# ---------------------------------------------------------------------------
# Public tool functions
# ---------------------------------------------------------------------------
def calculate_down_payment_power(
portfolio_value: float,
target_cities: Optional[list[str]] = None,
) -> dict:
"""
Calculates which housing markets a portfolio can fund at 20% down.
Scenarios:
full = portfolio_value (liquidate everything)
conservative = portfolio_value * 0.80 (keep 20% buffer)
safe = portfolio_value * 0.60 (maintain diversification)
Args:
portfolio_value: Total liquid portfolio value in USD.
target_cities: List of city names. Defaults to all 7 Austin ACTRIS markets.
Returns:
Dict with portfolio_value, down_payment_scenarios, markets list,
top_recommendation, mortgage_assumptions, data_source.
"""
full = portfolio_value
conservative = round(portfolio_value * 0.80, 2)
safe = round(portfolio_value * 0.60, 2)
if target_cities is None:
# Default: all 7 Austin ACTRIS markets (from _MOCK_SNAPSHOTS directly)
city_keys = _DEFAULT_AUSTIN_MARKETS
markets_data = [
(key, _MOCK_SNAPSHOTS[key])
for key in city_keys
if key in _MOCK_SNAPSHOTS
]
else:
markets_data = []
for city in target_cities:
display, data = _resolve_city_data_sync(city)
markets_data.append((city, data))
markets_out = []
affordable_markets = []
for city_ref, snap in markets_data:
price = snap.get("median_price") or snap.get("ListPrice") or 500_000
rent = snap.get("MedianRentMonthly") or snap.get("median_rent") or 0
area_name = snap.get("region") or snap.get("city") or city_ref
ds = snap.get("data_source", "estimate")
required_down_20 = round(price * 0.20, 2)
can_full = full >= required_down_20
can_conservative = conservative >= required_down_20
can_safe = safe >= required_down_20
monthly_payment = _monthly_payment(price)
rent_vs_buy_diff = round(monthly_payment - rent, 0) if rent else None
rent_vs_buy_verdict = (
f"Buying costs ${abs(rent_vs_buy_diff):,.0f}/mo "
f"{'more' if rent_vs_buy_diff > 0 else 'less'} than renting"
if rent_vs_buy_diff is not None else "Rental data unavailable"
)
# Simple break-even: closing costs ~3% + transaction costs ~6% = ~9% of price
# Break even = (9% of price) / monthly_savings_vs_rent
monthly_savings = -(rent_vs_buy_diff or 1)
if monthly_savings > 0 and rent_vs_buy_diff is not None:
break_even_years = round((price * 0.09) / (monthly_savings * 12), 1)
elif rent_vs_buy_diff is not None and rent_vs_buy_diff <= 0:
break_even_years = 0.0 # Already cheaper to buy
else:
break_even_years = None
entry = {
"area": area_name,
"city_ref": city_ref,
"median_price": price,
"required_down_20pct": required_down_20,
"can_afford_full": can_full,
"can_afford_conservative": can_conservative,
"can_afford_safe": can_safe,
"monthly_payment_estimate": int(monthly_payment),
"median_rent": rent,
"rent_vs_buy_monthly_diff": rent_vs_buy_diff,
"rent_vs_buy_verdict": rent_vs_buy_verdict,
"break_even_years": break_even_years,
"data_source": ds,
}
markets_out.append(entry)
if can_full:
affordable_markets.append(area_name)
# Build recommendation
max_home_price = round(portfolio_value / 0.20, 0)
affordable_conservatively = [
m for m in markets_out if m["can_afford_conservative"]
]
if not affordable_markets and not affordable_conservatively:
recommendation = (
f"Your ${portfolio_value:,.0f} portfolio covers 20% down on homes up to "
f"${max_home_price:,.0f}. None of the target markets fall below this threshold. "
"Consider markets in Caldwell County ($237,491) or increase savings before buying."
)
else:
reachable_names = [m["area"] for m in affordable_conservatively] or affordable_markets
recommendation = (
f"Your ${portfolio_value:,.0f} portfolio could fund a 20% down payment on "
f"homes up to ${max_home_price:,.0f}. "
+ (
f"Reachable markets (conservatively): {', '.join(reachable_names[:3])}."
if reachable_names else
f"Reachable with full liquidation: {', '.join(affordable_markets[:3])}."
)
)
return {
"portfolio_value": portfolio_value,
"down_payment_scenarios": {
"full": full,
"conservative": conservative,
"safe": safe,
},
"markets": markets_out,
"top_recommendation": recommendation,
"mortgage_assumptions": {
"rate": _MORTGAGE_RATE,
"term_years": 30,
"down_payment_pct": 20,
"disclaimer": "Rate is an estimate (6.95% 30yr fixed). Verify with lender.",
},
"data_source": (
"ACTRIS/Unlock MLS Jan 2026 (Austin areas) + Teleport API (other cities)"
),
}
async def calculate_job_offer_affordability(
offer_salary: float,
offer_city: str,
current_salary: float,
current_city: str,
) -> dict:
"""
Determines whether a job offer is a real raise in purchasing power terms.
Works for any two cities worldwide:
- Austin TX areas: uses ACTRIS COL index (real data)
- All other cities: uses Teleport API live data (or fallback)
Args:
offer_salary: Gross annual salary of the new offer, in USD.
offer_city: Destination city for the offer.
current_salary: Current gross annual salary, in USD.
current_city: Current city of residence.
Returns:
Full comparison dict including adjusted purchasing power, verdict,
break-even salary, state tax note, and housing cost comparison.
"""
# Fetch city data (async for Teleport; sync-cached for Austin)
async def _get_data(city: str) -> tuple[float, dict]:
if _is_austin_area(city):
city_key = _normalize_city(city)
snap = _MOCK_SNAPSHOTS.get(city_key, {})
col = _AUSTIN_COL_INDEX.get(city_key, 95.4)
return col, snap
data = await get_city_housing_data(city)
col = _col_index_for_city(city, data)
return col, data
current_col, current_data = await _get_data(current_city)
offer_col, offer_data = await _get_data(offer_city)
# Core purchasing power calculation
adjusted_offer = round(offer_salary * (current_col / offer_col), 2)
real_raise = round(adjusted_offer - current_salary, 2)
is_real_raise = real_raise > 0
pct_change = round((real_raise / current_salary), 4) if current_salary > 0 else 0.0
breakeven_salary = round(current_salary * (offer_col / current_col), 2)
current_city_display = current_data.get("city") or current_city
offer_city_display = offer_data.get("city") or offer_city
# Housing comparison
current_rent = (
current_data.get("MedianRentMonthly")
or current_data.get("median_rent")
or 0
)
offer_rent = (
offer_data.get("MedianRentMonthly")
or offer_data.get("median_rent")
or 0
)
rent_diff = round(offer_rent - current_rent, 0) if current_rent and offer_rent else None
# Verdict
if is_real_raise:
verdict = (
f"The {offer_city_display} offer is a REAL raise. "
f"${offer_salary:,.0f}/yr in {offer_city_display} is worth "
f"${adjusted_offer:,.0f} in {current_city_display} purchasing power terms — "
f"a genuine ${real_raise:,.0f} improvement ({pct_change * 100:.1f}%)."
)
else:
verdict = (
f"Despite looking like a ${offer_salary - current_salary:,.0f} raise, "
f"the {offer_city_display} offer is worth ${abs(real_raise):,.0f} LESS "
f"than your {current_city_display} salary in real purchasing power. "
f"You would need ${breakeven_salary:,.0f} in {offer_city_display} "
f"to match your current lifestyle."
)
tax_note = _state_tax_note(current_city, offer_city)
return {
"current_salary": current_salary,
"current_city": current_city_display,
"current_col_index": current_col,
"offer_salary": offer_salary,
"offer_city": offer_city_display,
"offer_col_index": offer_col,
"adjusted_offer_in_current_city_terms": adjusted_offer,
"real_raise_amount": real_raise,
"is_real_raise": is_real_raise,
"percentage_real_change": pct_change,
"breakeven_salary_needed": breakeven_salary,
"verdict": verdict,
"tax_note": tax_note,
"housing_comparison": {
"current_city_median_rent": current_rent,
"offer_city_median_rent": offer_rent,
"monthly_rent_difference": rent_diff,
},
"data_source": (
f"ACTRIS MLS (Austin areas) + Teleport API ({offer_city_display})"
),
"offer_data_source": offer_data.get("data_source", ""),
"current_data_source": current_data.get("data_source", ""),
}
async def get_portfolio_real_estate_summary(
target_cities: Optional[list[str]] = None,
) -> dict:
"""
Master function: reads live Ghostfolio portfolio then runs down payment analysis.
Steps:
1. Calls portfolio_analysis to get total portfolio value and top holdings
2. Passes total to calculate_down_payment_power() with target_cities
3. Returns combined result with quick plain-English answer
Args:
target_cities: Optional list of cities to analyze. Defaults to all Austin markets.
Returns:
Combined dict with portfolio_summary, down_payment_analysis, quick_answer.
"""
from agent.tools.portfolio import portfolio_analysis
portfolio_result = await portfolio_analysis()
portfolio_value = 0.0
top_holdings: list[str] = []
portfolio_error: Optional[str] = None
if portfolio_result.get("success"):
summary = portfolio_result.get("result", {}).get("summary", {})
portfolio_value = summary.get("total_current_value_usd", 0.0)
holdings = portfolio_result.get("result", {}).get("holdings", [])
top_holdings = [
h.get("symbol") or h.get("name", "")
for h in holdings[:5]
if h.get("symbol") or h.get("name")
]
else:
portfolio_error = portfolio_result.get("error", {}).get("message", "Unknown error")
down_payment_analysis = calculate_down_payment_power(portfolio_value, target_cities)
max_home_price = round(portfolio_value / 0.20, 0) if portfolio_value > 0 else 0
affordable = [
m for m in down_payment_analysis["markets"]
if m["can_afford_conservative"]
]
if portfolio_value == 0.0:
quick_answer = (
"Could not retrieve portfolio value. "
+ (f"Error: {portfolio_error}" if portfolio_error else "")
)
elif affordable:
names = [m["area"] for m in affordable[:3]]
quick_answer = (
f"Your ${portfolio_value:,.0f} portfolio could fund a 20% down payment "
f"on homes up to ${max_home_price:,.0f}. "
f"Reachable markets: {', '.join(names)}."
)
else:
quick_answer = (
f"Your ${portfolio_value:,.0f} portfolio covers 20% down on homes up to "
f"${max_home_price:,.0f}. Check Caldwell County at $237,491 — "
"most affordable in the Austin area."
)
return {
"portfolio_summary": {
"total_value": portfolio_value,
"top_holdings": top_holdings,
"portfolio_error": portfolio_error,
"allocation_note": (
"Liquidating portfolio would trigger capital gains taxes — "
"consult a financial advisor before using investments for a down payment."
),
},
"down_payment_analysis": down_payment_analysis,
"quick_answer": quick_answer,
}