""" dashboard.py — MoneyMaker live monitoring dashboard. Usage: python dashboard.py # starts on http://localhost:5001 python dashboard.py --port 5002 Auto-refreshes every 60 seconds. Shows: • Portfolio P&L + C25 benchmark • Open positions with live prices • Closed trades (win/loss) • Signal accuracy • Recent runner log tail """ import argparse import json import os import time from datetime import datetime, timezone from pathlib import Path import yfinance as yf from flask import Flask, render_template_string from db import get_conn, DB_TYPE from report import _c25_day_return, _unrealised_pnl, _realised_pnl, _total_fees, _signal_accuracy CAPITAL = 10_000 LOG_DIR = Path(os.getenv("LOG_DIR", str(Path(__file__).parent / "logs"))) REFRESH = 60 # seconds app = Flask(__name__) # ── HTML template ──────────────────────────────────────────────────────────── TEMPLATE = """ MoneyMaker Dashboard

📈 MoneyMaker

DB: {{ db_type }}  ·  Opdateret: {{ now }}  ·  Refresh om {{ refresh }}s
Net P&L
{{ "{:+,.0f}".format(net_pnl) }} kr
{{ "{:+.2f}%".format(net_pct) }}
Urealiseret
{{ "{:+,.0f}".format(unreal) }} kr
{{ open_count }} åben{{ 'e' if open_count != 1 else '' }} position{{ 'er' if open_count != 1 else '' }}
Realiseret
{{ "{:+,.0f}".format(realised) }} kr
Gebyrer: {{ "{:,.0f}".format(fees) }} kr
C25 i dag
{% if c25_ret is not none %}
{{ "{:+.2f}%".format(c25_ret) }}
vs benchmark: {{ "{:+.2f}%".format(vs_bench) if vs_bench is not none else "—" }}
{% else %}
—
Marked lukket?
{% endif %}
Signal accuracy
{% if sig.total_trades > 0 %}
{{ "{:.0f}%".format(sig.accuracy_pct) }}
{{ sig.correct }} / {{ sig.total_trades }} handler korrekte
{% else %}
—
Ingen lukkede handler
{% endif %}
Kapital
{{ "{:,.0f}".format(capital) }} kr
Kontant: {{ "{:,.0f}".format(cash) }} kr
📊 Åbne positioner
{% if positions %} {% for p in positions %} {% endfor %}
TickerAntalKøbtNuP&LÆndringStopTakeStatus
{{ p.ticker }} {{ "{:.0f}".format(p.shares) }} {{ "{:,.0f}".format(p.entry) }} {{ "{:,.0f}".format(p.last) }} {{ "{:+,.0f}".format(p.unreal) }} {{ "{:+.1f}%".format(p.pct) }} {{ "{:,.0f}".format(p.stop) }} {{ "{:,.0f}".format(p.take) }} {% if p.stop_hit %}🔴 STOP {% elif p.take_hit %}🟡 TAKE {% else %}⏳ HOLD{% endif %}
{% else %}

Ingen åbne positioner.

{% endif %}
🏁 Lukkede handler
{% if trades %} {% for t in trades %} {% endfor %}
TickerHandlingAntalKursTotalP&LSignalDato
{{ t.ticker }} {{ t.action.upper() }} {{ "{:.0f}".format(t.shares) }} {{ "{:,.0f}".format(t.price) }} {{ "{:,.0f}".format(t.total_dkk) }} {{ "{:+,.0f}".format(t.pnl_dkk) if t.pnl_dkk is not none else "—" }} {% if t.signal_correct == 1 %}✅ korrekt {% elif t.signal_correct == 0 %}❌ forkert {% else %}—{% endif %} {{ t.event_date }}
{% else %}

Ingen handler endnu.

{% endif %}
🔬 NLP signal pipeline
Analyserede signalerAlert-triggers (≥threshold)Gns. score
{{ sig.total }} {{ sig.alerts }} {{ "{:.3f}".format(sig.avg_score) }}
{% if log_tail %}
📋 Seneste log ({{ log_file }})
{{ log_tail }}
{% endif %}
MoneyMaker · {{ db_type }} · hjess-desktop · auto-refresh {{ refresh }}s
""" # ── Data helpers ───────────────────────────────────────────────────────────── def _latest_log_tail(lines: int = 40) -> tuple[str, str]: """Return (filename, last N lines) from most recent runner log.""" if not LOG_DIR.exists(): return "", "" logs = sorted(LOG_DIR.glob("runner_*.log"), reverse=True) if not logs: return "", "" path = logs[0] try: content = path.read_text(errors="replace") tail = "\n".join(content.splitlines()[-lines:]) return path.name, tail except Exception: return path.name, "" def _open_positions_live(db) -> list[dict]: """Fetch open positions with live yfinance prices.""" rows = db.execute("SELECT * FROM positions").fetchall() result = [] for p in rows: ticker = p["ticker"] from report import C25 yf_ticker = C25.get(ticker, {}).get("ticker_yahoo", ticker + ".CO") try: last = yf.Ticker(yf_ticker).fast_info.get("lastPrice") or p["entry_price"] except Exception: last = p["entry_price"] entry = float(p["entry_price"]) shares = float(p["shares"]) pct = (last - entry) / entry * 100 result.append({ "ticker": ticker, "shares": shares, "entry": entry, "last": last, "unreal": shares * (last - entry), "pct": pct, "stop": float(p["stop_loss"]), "take": float(p["take_profit"]), "stop_hit": last <= p["stop_loss"], "take_hit": last >= p["take_profit"], }) return result def _closed_trades(db, limit: int = 20) -> list[dict]: rows = db.execute(""" SELECT ticker, action, shares, price, total_dkk, pnl_dkk, signal_correct, event_date FROM position_events ORDER BY id DESC LIMIT ? """, (limit,)).fetchall() return [dict(r) for r in rows] # ── Routes ──────────────────────────────────────────────────────────────────── @app.route("/") def index(): db = get_conn() try: positions = _open_positions_live(db) trades = _closed_trades(db) unreal = sum(p["unreal"] for p in positions) invested = sum(p["shares"] * p["entry"] for p in positions) realised = _realised_pnl(db) fees = _total_fees(db) sig = _signal_accuracy(db) c25_ret = _c25_day_return() net_pnl = realised + unreal - fees net_pct = net_pnl / CAPITAL * 100 vs_bench = (net_pct - c25_ret) if c25_ret is not None else None cash = CAPITAL - invested log_file, log_tail = _latest_log_tail() now = datetime.now(timezone.utc).strftime("%d %b %Y %H:%M UTC") return render_template_string( TEMPLATE, now=now, db_type=DB_TYPE, refresh=REFRESH, capital=CAPITAL, cash=cash, unreal=unreal, realised=realised, fees=fees, net_pnl=net_pnl, net_pct=net_pct, c25_ret=c25_ret, vs_bench=vs_bench, positions=positions, open_count=len(positions), trades=trades, sig=type("Sig", (), sig)(), log_file=log_file, log_tail=log_tail, ) finally: db.close() @app.route("/health") def health(): return {"status": "ok", "db": DB_TYPE, "ts": time.time()} # ── Main ────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser(description="MoneyMaker Dashboard") parser.add_argument("--port", type=int, default=int(os.getenv("PORT", 5001))) parser.add_argument("--host", default=os.getenv("HOST", "0.0.0.0")) args = parser.parse_args() print(f"\n MoneyMaker Dashboard -> http://localhost:{args.port}\n") app.run(host=args.host, port=args.port, debug=False) if __name__ == "__main__": main()