| from fastapi import FastAPI, Request, Depends, Form, HTTPException, UploadFile, File |
| from fastapi.templating import Jinja2Templates |
| from fastapi.staticfiles import StaticFiles |
| from fastapi.responses import HTMLResponse, RedirectResponse |
| from sqlalchemy.orm import Session |
| from pathlib import Path |
| from datetime import date |
|
|
| from database import init_db, get_db |
| from models import UserProfile, Asset, Snapshot |
| from schemas import AssetCreate, AssetUpdate, SnapshotCreate |
| from fire_math import calculate_fire, projection_timeline, calculate_scenario |
| from csv_import import parse_bank_csv |
|
|
| app = FastAPI(title="FireTracker") |
|
|
| templates = Jinja2Templates(directory=Path(__file__).parent / "templates") |
| app.mount("/static", StaticFiles(directory=Path(__file__).parent / "static"), name="static") |
|
|
| @app.on_event("startup") |
| def startup(): |
| init_db() |
|
|
| def get_or_create_profile(db: Session) -> UserProfile: |
| """Get the single user profile, create if not exists.""" |
| profile = db.query(UserProfile).first() |
| if not profile: |
| profile = UserProfile(name="Samuel") |
| db.add(profile) |
| db.commit() |
| db.refresh(profile) |
| return profile |
|
|
| |
|
|
| @app.get("/", response_class=HTMLResponse) |
| def dashboard(request: Request, db: Session = Depends(get_db)): |
| profile = get_or_create_profile(db) |
| if profile.monthly_income == 0: |
| return RedirectResponse("/setup", status_code=303) |
| assets = db.query(Asset).all() |
| total_portfolio = sum(a.current_value for a in assets) |
|
|
| fire = calculate_fire( |
| monthly_income=profile.monthly_income, |
| monthly_expenses=profile.monthly_expenses, |
| total_portfolio=total_portfolio, |
| birth_year=profile.birth_year, |
| swr=profile.swr, |
| expected_return=profile.expected_return, |
| inflation=profile.inflation, |
| target_retirement_age=profile.target_retirement_age, |
| ) |
|
|
| timeline = projection_timeline( |
| total_portfolio=total_portfolio, |
| monthly_savings=fire.monthly_savings, |
| annual_return=fire.real_return, |
| years=30, |
| ) |
|
|
| |
| asset_groups = {} |
| for a in assets: |
| asset_groups.setdefault(a.asset_type, []).append(a) |
|
|
| return templates.TemplateResponse("dashboard.html", { |
| "request": request, |
| "profile": profile, |
| "fire": fire, |
| "assets": assets, |
| "asset_groups": asset_groups, |
| "total_portfolio": total_portfolio, |
| "timeline": timeline, |
| "page": "dashboard", |
| }) |
|
|
| @app.get("/assets", response_class=HTMLResponse) |
| def assets_page(request: Request, db: Session = Depends(get_db)): |
| assets = db.query(Asset).all() |
| return templates.TemplateResponse("assets.html", { |
| "request": request, |
| "assets": assets, |
| "page": "assets", |
| "asset_types": ["depot", "immobilie", "cash", "crypto", "pension", "other"], |
| }) |
|
|
| @app.get("/settings", response_class=HTMLResponse) |
| def settings_page(request: Request, db: Session = Depends(get_db)): |
| profile = get_or_create_profile(db) |
| return templates.TemplateResponse("settings.html", { |
| "request": request, |
| "profile": profile, |
| "page": "settings", |
| }) |
|
|
| @app.get("/setup", response_class=HTMLResponse) |
| def setup_page(request: Request, db: Session = Depends(get_db)): |
| profile = get_or_create_profile(db) |
| return templates.TemplateResponse("setup.html", { |
| "request": request, |
| "profile": profile, |
| "page": "setup", |
| }) |
|
|
| |
|
|
| @app.post("/api/profile", response_class=HTMLResponse) |
| def update_profile( |
| request: Request, |
| monthly_income: float = Form(...), |
| monthly_expenses: float = Form(...), |
| birth_year: int = Form(...), |
| target_retirement_age: int = Form(67), |
| swr: float = Form(0.04), |
| expected_return: float = Form(0.07), |
| inflation: float = Form(0.02), |
| db: Session = Depends(get_db), |
| ): |
| profile = get_or_create_profile(db) |
| profile.monthly_income = monthly_income |
| profile.monthly_expenses = monthly_expenses |
| profile.birth_year = birth_year |
| profile.target_retirement_age = target_retirement_age |
| profile.swr = swr |
| profile.expected_return = expected_return |
| profile.inflation = inflation |
| db.commit() |
| return RedirectResponse("/", status_code=303) |
|
|
| @app.post("/api/setup", response_class=HTMLResponse) |
| def complete_setup( |
| request: Request, |
| monthly_income: float = Form(...), |
| monthly_expenses: float = Form(...), |
| birth_year: int = Form(...), |
| db: Session = Depends(get_db), |
| ): |
| """Quick setup — just the 3 essential numbers.""" |
| profile = get_or_create_profile(db) |
| profile.monthly_income = monthly_income |
| profile.monthly_expenses = monthly_expenses |
| profile.birth_year = birth_year |
| db.commit() |
| return RedirectResponse("/", status_code=303) |
|
|
| @app.post("/api/assets") |
| def create_asset( |
| request: Request, |
| name: str = Form(...), |
| asset_type: str = Form(...), |
| current_value: float = Form(0), |
| monthly_contribution: float = Form(0), |
| notes: str = Form(""), |
| db: Session = Depends(get_db), |
| ): |
| asset = Asset( |
| name=name, |
| asset_type=asset_type, |
| current_value=current_value, |
| monthly_contribution=monthly_contribution, |
| notes=notes, |
| ) |
| db.add(asset) |
| db.commit() |
| |
| snapshot = Snapshot(asset_id=asset.id, value=current_value, date=date.today()) |
| db.add(snapshot) |
| db.commit() |
| return RedirectResponse("/assets", status_code=303) |
|
|
| @app.put("/api/assets/{asset_id}", response_class=HTMLResponse) |
| def update_asset( |
| asset_id: int, |
| request: Request, |
| name: str = Form(...), |
| asset_type: str = Form(...), |
| current_value: float = Form(0), |
| monthly_contribution: float = Form(0), |
| notes: str = Form(""), |
| db: Session = Depends(get_db), |
| ): |
| asset = db.get(Asset, asset_id) |
| if not asset: |
| raise HTTPException(404) |
| old_value = asset.current_value |
| asset.name = name |
| asset.asset_type = asset_type |
| asset.current_value = current_value |
| asset.monthly_contribution = monthly_contribution |
| asset.notes = notes |
| db.commit() |
| |
| if current_value != old_value: |
| snapshot = Snapshot(asset_id=asset.id, value=current_value, date=date.today()) |
| db.add(snapshot) |
| db.commit() |
| |
| assets = db.query(Asset).all() |
| return templates.TemplateResponse("partials/asset_list.html", { |
| "request": request, |
| "assets": assets, |
| "asset_types": ["depot", "immobilie", "cash", "crypto", "pension", "other"], |
| }) |
|
|
| @app.delete("/api/assets/{asset_id}", response_class=HTMLResponse) |
| def delete_asset( |
| asset_id: int, |
| request: Request, |
| db: Session = Depends(get_db), |
| ): |
| asset = db.get(Asset, asset_id) |
| if not asset: |
| raise HTTPException(404) |
| db.delete(asset) |
| db.commit() |
| assets = db.query(Asset).all() |
| return templates.TemplateResponse("partials/asset_list.html", { |
| "request": request, |
| "assets": assets, |
| "asset_types": ["depot", "immobilie", "cash", "crypto", "pension", "other"], |
| }) |
|
|
| @app.post("/api/snapshots") |
| def add_snapshot( |
| asset_id: int = Form(...), |
| value: float = Form(...), |
| snapshot_date: str = Form(None), |
| db: Session = Depends(get_db), |
| ): |
| asset = db.get(Asset, asset_id) |
| if not asset: |
| raise HTTPException(404) |
| d = date.fromisoformat(snapshot_date) if snapshot_date else date.today() |
| snapshot = Snapshot(asset_id=asset_id, value=value, date=d) |
| db.add(snapshot) |
| asset.current_value = value |
| db.commit() |
| return RedirectResponse("/assets", status_code=303) |
|
|
| |
|
|
| @app.get("/api/timeline") |
| def get_timeline(db: Session = Depends(get_db)): |
| profile = get_or_create_profile(db) |
| assets = db.query(Asset).all() |
| total = sum(a.current_value for a in assets) |
| fire = calculate_fire( |
| monthly_income=profile.monthly_income, |
| monthly_expenses=profile.monthly_expenses, |
| total_portfolio=total, |
| birth_year=profile.birth_year, |
| swr=profile.swr, |
| expected_return=profile.expected_return, |
| inflation=profile.inflation, |
| ) |
| timeline = projection_timeline(total, fire.monthly_savings, fire.real_return, 30) |
| return {"timeline": timeline, "fi_number": fire.fi_number} |
|
|
| @app.get("/api/scenario") |
| def get_scenario( |
| extra_savings: float = 0, |
| adjusted_expenses: float = None, |
| db: Session = Depends(get_db), |
| ): |
| profile = get_or_create_profile(db) |
| assets = db.query(Asset).all() |
| total = sum(a.current_value for a in assets) |
|
|
| base = calculate_fire( |
| monthly_income=profile.monthly_income, |
| monthly_expenses=profile.monthly_expenses, |
| total_portfolio=total, |
| birth_year=profile.birth_year, |
| swr=profile.swr, |
| expected_return=profile.expected_return, |
| inflation=profile.inflation, |
| ) |
|
|
| scenario = calculate_scenario( |
| base=base, |
| monthly_income=profile.monthly_income, |
| monthly_expenses=profile.monthly_expenses, |
| total_portfolio=total, |
| birth_year=profile.birth_year, |
| extra_monthly_savings=extra_savings, |
| adjusted_expenses=adjusted_expenses, |
| swr=profile.swr, |
| expected_return=profile.expected_return, |
| inflation=profile.inflation, |
| ) |
|
|
| return { |
| "base_years": base.years_to_fi, |
| "scenario_years": scenario.years_to_fi, |
| "base_fi_date": str(base.fi_date) if base.fi_date else None, |
| "scenario_fi_date": str(scenario.fi_date) if scenario.fi_date else None, |
| "diff_years": round(base.years_to_fi - scenario.years_to_fi, 1) if base.years_to_fi and scenario.years_to_fi else None, |
| } |
|
|
| @app.get("/import", response_class=HTMLResponse) |
| def import_page(request: Request, db: Session = Depends(get_db)): |
| return templates.TemplateResponse("import.html", { |
| "request": request, |
| "page": "import", |
| }) |
|
|
| @app.post("/api/import/csv") |
| async def import_csv( |
| request: Request, |
| file: UploadFile = File(...), |
| bank_format: str = Form("auto"), |
| db: Session = Depends(get_db), |
| ): |
| """Parse bank CSV and return categorized transactions.""" |
|
|
| content = await file.read() |
| result = parse_bank_csv(content) |
|
|
| return templates.TemplateResponse("import.html", { |
| "request": request, |
| "page": "import", |
| "transactions": result["transactions"], |
| "total_income": round(result["total_income"], 2), |
| "total_expenses": round(result["total_expenses"], 2), |
| "net": round(result["net"], 2), |
| "count": result["count"], |
| "monthly_estimate": round(result["monthly_estimate"], 2), |
| }) |
|
|
| @app.get("/api/export") |
| def export_data(db: Session = Depends(get_db)): |
| """Export all data as JSON for backup/portability.""" |
| profile = get_or_create_profile(db) |
| assets = db.query(Asset).all() |
| snapshots = db.query(Snapshot).all() |
| return { |
| "profile": { |
| "monthly_income": profile.monthly_income, |
| "monthly_expenses": profile.monthly_expenses, |
| "birth_year": profile.birth_year, |
| "target_retirement_age": profile.target_retirement_age, |
| "swr": profile.swr, |
| "expected_return": profile.expected_return, |
| "inflation": profile.inflation, |
| }, |
| "assets": [{"name": a.name, "type": a.asset_type, "value": a.current_value, "contribution": a.monthly_contribution, "notes": a.notes} for a in assets], |
| "snapshots": [{"asset_id": s.asset_id, "date": str(s.date), "value": s.value} for s in snapshots], |
| } |
|
|
