Viktor Barzin
e72fd22a17
The Monte Carlo used to compare jurisdictions at a flat London-equivalent spend, which silently overstated the cost-of-living for any move to a cheaper region. Now every cross-jurisdiction simulation auto-scales spending_gbp by the real Numbeo/Expatistan ratio between the user's baseline city and the target city. Architecture: - fire_planner/col/baseline.py — 22 cities with headline Numbeo data (source URLs + snapshot dates embedded) — fallback when scraper fails - col/numbeo.py + col/expatistan.py — httpx async scrapers, regex-parsed, polite 1.1s rate-limit, EUR/USD anchored - col/cache.py — PG-backed cache (col_snapshot table, 1-year TTL) - col/service.py — sync compute_col_ratio() for the simulator; async lookup_city_cached() with source reconciliation for the refresh CronJob - alembic 0005 — col_snapshot table, UNIQUE(city_slug, source_name) Simulator wiring: - SimulateRequest gains col_auto_adjust=True (default), col_baseline_city, col_target_city. Defaults pick the jurisdiction's representative city. - _resolve_col_adjustment scales spending_gbp before path-building. - SimulateResult surfaces col_multiplier_applied + col_adjusted_spending_gbp. CLIs: - python -m fire_planner col-seed — loads BASELINES into col_snapshot (post-migration seed step) - python -m fire_planner col-refresh-stale --within-days 7 — used by the weekly fire-planner-col-refresh CronJob 268 tests pass. Mypy strict + ruff clean. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
367 lines
14 KiB
Python
367 lines
14 KiB
Python
"""click CLI entrypoint.
|
|
|
|
Sub-commands:
|
|
- migrate — alembic upgrade head
|
|
- ingest [wealthfolio] — pull from wealthfolio_sync PG mirror into account_snapshot
|
|
- simulate — run a single scenario, pretty-print
|
|
- recompute-all — run the 120-scenario Cartesian, persist all
|
|
- serve — run the FastAPI on-demand /recompute server
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
import click
|
|
import numpy as np
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
from fire_planner.db import create_engine_from_env, make_session_factory
|
|
from fire_planner.glide_path import get as get_glide
|
|
from fire_planner.ingest.wealthfolio import upsert_snapshots
|
|
from fire_planner.ingest.wealthfolio_pg import (
|
|
create_wf_sync_engine_from_env,
|
|
read_account_snapshots_from_pg,
|
|
)
|
|
from fire_planner.reporters.cli import format_scenario
|
|
from fire_planner.reporters.pg import write_run
|
|
from fire_planner.returns.bootstrap import block_bootstrap
|
|
from fire_planner.returns.shiller import load_from_csv, synthetic_returns
|
|
from fire_planner.scenarios import (
|
|
ScenarioSpec,
|
|
build_regime_schedule,
|
|
build_strategy,
|
|
cartesian_scenarios,
|
|
)
|
|
from fire_planner.simulator import simulate
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@click.group()
|
|
def cli() -> None:
|
|
logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
|
|
|
|
|
|
@cli.command()
|
|
def migrate() -> None:
|
|
"""Run `alembic upgrade head`."""
|
|
rc = subprocess.run(["alembic", "upgrade", "head"], check=False)
|
|
sys.exit(rc.returncode)
|
|
|
|
|
|
@cli.command("col-seed")
|
|
@click.option("--ttl-days",
|
|
type=int,
|
|
default=365,
|
|
help="Cache TTL in days (default 365 — matches Viktor's 1y choice).")
|
|
def col_seed(ttl_days: int) -> None:
|
|
"""Seed `col_snapshot` from baseline.py BASELINES.
|
|
|
|
Idempotent — uses upsert on (city_slug, source_name). Run once after
|
|
the alembic migration creates the table. Subsequent live-scrape
|
|
refreshes (Phase 3 CronJob) supersede these rows; the baseline
|
|
fallback remains as a last-resort source.
|
|
"""
|
|
asyncio.run(_col_seed(ttl_days))
|
|
|
|
|
|
async def _col_seed(ttl_days: int) -> None:
|
|
from fire_planner.col.baseline import BASELINES
|
|
from fire_planner.col.cache import upsert as col_upsert
|
|
|
|
engine = create_engine_from_env()
|
|
factory = make_session_factory(engine)
|
|
try:
|
|
async with factory() as sess:
|
|
for slug, idx in BASELINES.items():
|
|
# Tag the source as `baseline` rather than `numbeo` so a
|
|
# later live scrape (source_name='numbeo') doesn't conflict
|
|
# on the (city_slug, source_name) unique constraint.
|
|
tagged = idx.model_copy(
|
|
update={"source": idx.source.model_copy(update={"name": "baseline"})}
|
|
)
|
|
await col_upsert(sess, tagged, ttl_days=ttl_days)
|
|
click.echo(f" seeded {slug:20s} total={idx.total_single_with_rent_gbp} GBP")
|
|
finally:
|
|
await engine.dispose()
|
|
click.echo(f"\ncol-seed: {len(BASELINES)} cities upserted (ttl_days={ttl_days}).")
|
|
|
|
|
|
@cli.command("col-refresh-stale")
|
|
@click.option("--within-days",
|
|
type=int,
|
|
default=7,
|
|
help="Refresh rows whose expires_at is within this many days.")
|
|
@click.option("--ttl-days",
|
|
type=int,
|
|
default=365,
|
|
help="TTL for re-written rows (default 365).")
|
|
def col_refresh_stale(within_days: int, ttl_days: int) -> None:
|
|
"""Re-scrape COL rows that are within `within_days` of expiry.
|
|
|
|
Designed for the weekly CronJob. Walks every distinct city_slug in
|
|
`col_snapshot` whose newest row will expire within the window,
|
|
calls Numbeo+Expatistan via `service.lookup_city_cached`, which
|
|
upserts the result. Idempotent — no-op for fresh rows.
|
|
"""
|
|
asyncio.run(_col_refresh_stale(within_days, ttl_days))
|
|
|
|
|
|
async def _col_refresh_stale(within_days: int, ttl_days: int) -> None:
|
|
from sqlalchemy import select, text
|
|
|
|
from fire_planner.col.service import lookup_city_cached
|
|
from fire_planner.db import ColSnapshot
|
|
|
|
engine = create_engine_from_env()
|
|
factory = make_session_factory(engine)
|
|
threshold = f"NOW() + INTERVAL '{int(within_days)} days'"
|
|
refreshed = 0
|
|
failed = 0
|
|
try:
|
|
async with factory() as sess:
|
|
# Find distinct city_slug whose freshest row expires within window.
|
|
stmt = (
|
|
select(ColSnapshot.city_slug, ColSnapshot.country)
|
|
.distinct()
|
|
.where(text(f"expires_at <= {threshold}"))
|
|
)
|
|
rows = (await sess.execute(stmt)).all()
|
|
click.echo(f"col-refresh-stale: {len(rows)} city(ies) need refresh "
|
|
f"(within_days={within_days})")
|
|
for slug, country in rows:
|
|
try:
|
|
# lookup_city_cached upserts on cache miss, which is
|
|
# what "stale" means here — read_fresh returns None.
|
|
idx = await lookup_city_cached(sess, slug, country=country or "")
|
|
click.echo(f" refreshed {slug:20s} → {idx.source.name:10s} "
|
|
f"total={idx.total_single_with_rent_gbp}")
|
|
refreshed += 1
|
|
except Exception as e: # broad — log and continue per-city
|
|
click.echo(f" FAILED {slug}: {e}", err=True)
|
|
failed += 1
|
|
finally:
|
|
await engine.dispose()
|
|
click.echo(f"\ncol-refresh-stale done: refreshed={refreshed} failed={failed} "
|
|
f"ttl_days={ttl_days}")
|
|
|
|
|
|
@cli.command("ingest")
|
|
@click.option("--source",
|
|
type=click.Choice(["wealthfolio"]),
|
|
default="wealthfolio",
|
|
help="Data source — currently only wealthfolio is wired.")
|
|
@click.option("--as-of",
|
|
type=click.DateTime(formats=["%Y-%m-%d"]),
|
|
default=None,
|
|
help="Valuation date to read; defaults to MAX(valuation_date) in the mirror.")
|
|
def ingest(source: str, as_of: date | None) -> None:
|
|
"""Pull external state into fire_planner.account_snapshot."""
|
|
if source == "wealthfolio":
|
|
asyncio.run(_ingest_wealthfolio(as_of))
|
|
|
|
|
|
async def _ingest_wealthfolio(as_of: date | None) -> None:
|
|
"""Read account snapshots from wealthfolio_sync PG mirror, upsert."""
|
|
wf_engine = create_wf_sync_engine_from_env()
|
|
try:
|
|
wf_factory = async_sessionmaker(wf_engine, expire_on_commit=False)
|
|
async with wf_factory() as wf_sess:
|
|
rows = await read_account_snapshots_from_pg(wf_sess, as_of=as_of)
|
|
finally:
|
|
await wf_engine.dispose()
|
|
|
|
if not rows:
|
|
click.echo(
|
|
"warning: no rows read — wealthfolio_sync mirror is empty or "
|
|
"no rows on the requested date",
|
|
err=True,
|
|
)
|
|
|
|
engine = create_engine_from_env()
|
|
factory = make_session_factory(engine)
|
|
try:
|
|
async with factory() as sess:
|
|
n = await upsert_snapshots(sess, rows)
|
|
await sess.commit()
|
|
click.echo(f"wealthfolio ingest: {n} rows upserted")
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
def _build_paths(seed: int, n_paths: int, n_years: int, returns_csv: Path | None) -> np.ndarray:
|
|
"""Load returns from CSV (production) or synthetic (smoke tests)."""
|
|
if returns_csv and returns_csv.exists():
|
|
bundle = load_from_csv(returns_csv)
|
|
else:
|
|
bundle = synthetic_returns(seed=42)
|
|
rng = np.random.default_rng(seed)
|
|
return block_bootstrap(bundle, n_paths=n_paths, n_years=n_years, block_size=5, rng=rng)
|
|
|
|
|
|
@cli.command("simulate")
|
|
@click.option("--scenario",
|
|
required=True,
|
|
help="external_id, e.g. cyprus-vpw-leave-y3-glide-rising")
|
|
@click.option("--n-paths", type=int, default=10_000)
|
|
@click.option("--horizon", type=int, default=60)
|
|
@click.option("--spending", type=float, default=100_000.0)
|
|
@click.option("--nw-seed", type=float, default=1_000_000.0)
|
|
@click.option("--savings", type=float, default=0.0)
|
|
@click.option("--floor",
|
|
type=float,
|
|
default=None,
|
|
help="Real-GBP floor for vpw_floor strategy (e.g. 40000).")
|
|
@click.option("--returns-csv", type=click.Path(path_type=Path), default=None)
|
|
@click.option("--seed", type=int, default=42)
|
|
@click.option("--write-db/--no-write-db", default=False, help="Persist results to fire_planner DB.")
|
|
def simulate_cmd(
|
|
scenario: str,
|
|
n_paths: int,
|
|
horizon: int,
|
|
spending: float,
|
|
nw_seed: float,
|
|
savings: float,
|
|
floor: float | None,
|
|
returns_csv: Path | None,
|
|
seed: int,
|
|
write_db: bool,
|
|
) -> None:
|
|
"""Run one scenario by external_id and pretty-print the result."""
|
|
parts = scenario.split("-")
|
|
if len(parts) < 6 or parts[2] != "leave" or parts[4] != "glide":
|
|
raise click.UsageError(f"bad scenario id: {scenario!r} "
|
|
"(expected jurisdiction-strategy-leave-yN-glide-NAME)")
|
|
jurisdiction = parts[0]
|
|
# strategy may include underscore (e.g. guyton_klinger), so rebuild
|
|
strategy_end = scenario.index("-leave-")
|
|
strategy_name = scenario[len(jurisdiction) + 1:strategy_end]
|
|
leave_year = int(parts[parts.index("leave") + 1].lstrip("y"))
|
|
glide_name = scenario.split("-glide-")[1]
|
|
|
|
spec = ScenarioSpec(
|
|
jurisdiction=jurisdiction,
|
|
strategy=strategy_name,
|
|
leave_uk_year=leave_year,
|
|
glide_path=glide_name,
|
|
spending_gbp=Decimal(str(spending)),
|
|
nw_seed_gbp=Decimal(str(nw_seed)),
|
|
horizon_years=horizon,
|
|
savings_per_year_gbp=Decimal(str(savings)),
|
|
)
|
|
paths = _build_paths(seed, n_paths, horizon, returns_csv)
|
|
annual_savings = (np.full(horizon, savings, dtype=np.float64) if savings else None)
|
|
started = time.perf_counter()
|
|
result = simulate(
|
|
paths=paths,
|
|
initial_portfolio=nw_seed,
|
|
spending_target=spending,
|
|
glide=get_glide(glide_name),
|
|
strategy=build_strategy(strategy_name, floor=floor),
|
|
regime=build_regime_schedule(jurisdiction, leave_year),
|
|
horizon_years=horizon,
|
|
annual_savings=annual_savings,
|
|
)
|
|
elapsed = time.perf_counter() - started
|
|
click.echo(format_scenario(spec, result))
|
|
if write_db:
|
|
asyncio.run(_persist(spec, result, seed=seed, elapsed_seconds=elapsed))
|
|
click.echo(f"simulate: elapsed={elapsed:.2f}s")
|
|
|
|
|
|
async def _persist(spec: ScenarioSpec, result: object, *, seed: int,
|
|
elapsed_seconds: float) -> None:
|
|
engine = create_engine_from_env()
|
|
factory = make_session_factory(engine)
|
|
try:
|
|
async with factory() as sess:
|
|
from fire_planner.simulator import SimulationResult # local to avoid cycle
|
|
assert isinstance(result, SimulationResult)
|
|
await write_run(sess, spec, result, seed=seed, elapsed_seconds=elapsed_seconds)
|
|
await sess.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@cli.command("recompute-all")
|
|
@click.option("--n-paths", type=int, default=10_000)
|
|
@click.option("--horizon", type=int, default=60)
|
|
@click.option("--spending", type=float, default=100_000.0)
|
|
@click.option("--nw-seed", type=float, default=1_000_000.0)
|
|
@click.option("--savings", type=float, default=0.0)
|
|
@click.option("--floor",
|
|
type=float,
|
|
default=None,
|
|
help="Real-GBP floor — applied to vpw_floor scenarios in the sweep.")
|
|
@click.option("--returns-csv", type=click.Path(path_type=Path), default=None)
|
|
@click.option("--seed", type=int, default=42)
|
|
def recompute_all(n_paths: int, horizon: int, spending: float, nw_seed: float, savings: float,
|
|
floor: float | None, returns_csv: Path | None, seed: int) -> None:
|
|
"""Run the full Cartesian (default 120 scenarios) and persist."""
|
|
asyncio.run(
|
|
_recompute_all(n_paths, horizon, spending, nw_seed, savings, floor, returns_csv, seed))
|
|
|
|
|
|
async def _recompute_all(
|
|
n_paths: int,
|
|
horizon: int,
|
|
spending: float,
|
|
nw_seed: float,
|
|
savings: float,
|
|
floor: float | None,
|
|
returns_csv: Path | None,
|
|
seed: int,
|
|
) -> None:
|
|
paths = _build_paths(seed, n_paths, horizon, returns_csv)
|
|
specs = cartesian_scenarios(
|
|
spending_gbp=Decimal(str(spending)),
|
|
nw_seed_gbp=Decimal(str(nw_seed)),
|
|
savings_per_year_gbp=Decimal(str(savings)),
|
|
horizon_years=horizon,
|
|
)
|
|
annual_savings = (np.full(horizon, savings, dtype=np.float64) if savings else None)
|
|
engine = create_engine_from_env()
|
|
factory = make_session_factory(engine)
|
|
successes = 0
|
|
try:
|
|
async with factory() as sess:
|
|
for spec in specs:
|
|
started = time.perf_counter()
|
|
result = simulate(
|
|
paths=paths,
|
|
initial_portfolio=nw_seed,
|
|
spending_target=spending,
|
|
glide=get_glide(spec.glide_path),
|
|
strategy=build_strategy(spec.strategy, floor=floor),
|
|
regime=build_regime_schedule(spec.jurisdiction, spec.leave_uk_year),
|
|
horizon_years=horizon,
|
|
annual_savings=annual_savings,
|
|
)
|
|
elapsed = time.perf_counter() - started
|
|
await write_run(sess, spec, result, seed=seed, elapsed_seconds=elapsed)
|
|
successes += 1
|
|
click.echo(f"{spec.external_id}: success={result.success_rate*100:.1f}% "
|
|
f"elapsed={elapsed:.2f}s")
|
|
await sess.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
click.echo(f"recompute-all done: {successes}/{len(specs)} scenarios written")
|
|
|
|
|
|
@cli.command()
|
|
def serve() -> None:
|
|
"""Run the FastAPI on-demand /recompute server."""
|
|
import uvicorn
|
|
uvicorn.run("fire_planner.app:app", host="0.0.0.0", port=8080)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli()
|