""" 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 - Equity curve, win/loss, position charts (Apache ECharts) - 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, jsonify, 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

MoneyMaker

{{ db_type }}
{{ now }}
auto-refresh {{ refresh }}s
Net P&L
{{ "{:+,.0f}".format(net_pnl) }}
{{ "{:+.2f}%".format(net_pct) }} af kapital
Urealiseret
{{ "{:+,.0f}".format(unreal) }}
{{ open_count }} position{{ 'er' if open_count != 1 else '' }}
Realiseret
{{ "{:+,.0f}".format(realised) }}
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
{% else %}
—
ingen handler endnu
{% endif %}
Kapital
{{ "{:,.0f}".format(capital) }}
kontant: {{ "{:,.0f}".format(cash) }} kr
Equity curve
Win / Loss
Positioner P&L
Aben positioner {{ open_count }}
{% if positions %} {% for p in positions %} {% endfor %}
TickerAntal KobNu P&LAendring StopTakeStatus
{{ 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 abne positioner.
{% endif %}
Lukkede handler {{ trades|length }}
{% if trades %} {% for t in trades %} {% endfor %}
TickerTypeAntalKursTotalP&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 signaler Alert-triggers Gns. score Korrekte handler
{{ sig.total }} {{ sig.alerts }} {{ "{:.3f}".format(sig.avg_score) }} {{ sig.correct }} / {{ sig.total_trades }}
C25 Signal Board  ·  7 dage {{ signal_board|length }}
{% if signal_board %} {% for s in signal_board %} {% endfor %}
Selskab Pris Analytiker Nyheder Dækning Signalstyrke Vurdering
{{ s.ticker }} {{ s.name }} {% if s.alerts %}{{ s.alerts }} alert{% endif %} {{ s.price }} kr {{ s.rec_clean }} {{ s.rec_count }} {{ s.sent_text }} {{ s.n_articles }} art · {{ s.n_sources }} src {% if s.n_sources < 3 %} lav{% endif %}
{{ s.signal_str }}
{{ s.agree_text }}
{% else %}
Ingen signaler de seneste 7 dage — kør: make
{% endif %}
{% if log_tail %}
Runner log {{ log_file }} 
{{ log_tail }}
{% endif %}
MoneyMaker  ·  {{ db_type }}  ·  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] def _equity_curve(db) -> list[list]: """Cumulative realised P&L per day from position_events (sell events only).""" rows = db.execute(""" SELECT event_date, SUM(COALESCE(pnl_dkk, 0)) as daily_pnl FROM position_events WHERE action = 'sell' GROUP BY event_date ORDER BY event_date """).fetchall() cumulative = 0.0 result = [] for r in rows: cumulative += float(r["daily_pnl"] or 0) result.append([r["event_date"], round(cumulative, 0)]) if not result: from datetime import date result = [[str(date.today()), 0.0]] return result def _winloss_data(db) -> dict: """Win/loss/open counts for donut chart.""" rows = db.execute(""" SELECT signal_correct, COUNT(*) as cnt FROM position_events WHERE action = 'sell' GROUP BY signal_correct """).fetchall() counts = {r["signal_correct"]: r["cnt"] for r in rows} return { "win": counts.get(1, 0), "loss": counts.get(0, 0), "open": counts.get(None, 0), "total": sum(counts.values()), } def _signal_board_data(db) -> list[dict]: """Build signal board rows from DB + Yahoo Finance analyst consensus.""" from signals import company_stats, analyst_rec, C25 rows = company_stats(db, days=7) result = [] for r in rows: ticker = r["ticker"] company = C25.get(ticker, {}) avg_s = r["avg_sentiment"] or 0 # Sentiment badge if avg_s > 0.1: sent_class, sent_text = "badge-green", "Positiv" elif avg_s < -0.1: sent_class, sent_text = "badge-red", "Negativ" else: sent_class, sent_text = "badge-gray", "Neutral" # Analyst rec badge (signals.py caches per-process) rec = analyst_rec(ticker) if rec["mean"] is None: rec_class, rec_clean = "badge-gray", "Ukendt" elif rec["mean"] <= 1.5: rec_class, rec_clean = "badge-green", "Stærk KØB" elif rec["mean"] <= 2.5: rec_class, rec_clean = "badge-green", "KØB" elif rec["mean"] <= 3.5: rec_class, rec_clean = "badge-yellow", "HOLD" elif rec["mean"] <= 4.5: rec_class, rec_clean = "badge-red", "SÆLG" else: rec_class, rec_clean = "badge-red", "Stærk SÆLG" # Agreement between news and analysts news_bull = avg_s > 0.1 news_bear = avg_s < -0.1 ana_bull = rec["mean"] is not None and rec["mean"] <= 2.5 ana_bear = rec["mean"] is not None and rec["mean"] >= 3.5 if (news_bull and ana_bull) or (news_bear and ana_bear): agree_class, agree_text = "badge-green", "Enige" elif (news_bull and ana_bear) or (news_bear and ana_bull): agree_class, agree_text = "badge-red", "Uenige" else: agree_class, agree_text = "badge-gray", "—" # Signal strength bar (scale 0–0.5 → 0–100%) sig = r.get("max_signal") or 0 signal_pct = min(int(sig * 200), 100) if sig >= 0.5: sig_color = "var(--green)" elif sig >= 0.2: sig_color = "var(--yellow)" elif sig > 0: sig_color = "var(--accent)" else: sig_color = "var(--border)" result.append({ "ticker": ticker, "name": company.get("name", ticker), "price": company.get("price_dkk_approx", "?"), "rec_class": rec_class, "rec_clean": rec_clean, "rec_count": rec["count"], "sent_class": sent_class, "sent_text": sent_text, "n_articles": r["mention_articles"], "n_sources": int(r["avg_sources"] or 0), "signal": sig, "signal_str": f"{sig:.2f}", "signal_pct": signal_pct, "sig_color": sig_color, "agree_class": agree_class, "agree_text": agree_text, "alerts": int(r.get("alert_count") or 0), }) return result # ── 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") equity = _equity_curve(db) winloss = _winloss_data(db) pos_bars = [[p["ticker"], round(p["unreal"], 0)] for p in positions] or [["—", 0]] try: signal_board = _signal_board_data(db) except Exception as exc: signal_board = [] app.logger.warning("signal_board error: %s", exc) 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, equity_json=json.dumps(equity), winloss_json=json.dumps(winloss), positions_json=json.dumps(pos_bars), signal_board=signal_board, ) 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() from db import init_schema init_schema() print(f"\n MoneyMaker Dashboard -> http://localhost:{args.port}\n") app.run(host=args.host, port=args.port, debug=False) if __name__ == "__main__": main()