diff --git a/src/strategy_pythagoras/strategy_pythagoras/frontend/data.py b/src/strategy_pythagoras/strategy_pythagoras/frontend/data.py new file mode 100644 index 0000000..ed88a40 --- /dev/null +++ b/src/strategy_pythagoras/strategy_pythagoras/frontend/data.py @@ -0,0 +1,215 @@ +"""Paper-trading data access functions for the strategy_pythagoras dashboard. + +Reads exclusively from strategy_pythagoras_paper.db (paper_trading_* tables) +per il paper-trading; le funzioni dedicate ai winner Pythagoras leggono +invece dal runs.db del core GA (env ``GA_DB_PATH``), default +``state/strategy_pythagoras.db`` via env ``STRATEGY_PYTHAGORAS_DB_PATH`` quando +si vuole usare una sotto-tabella locale. +""" + +from __future__ import annotations + +import json +import sqlite3 +import time +from pathlib import Path +from typing import Any + +import pandas as pd # type: ignore[import-untyped] + + +def _paper_conn(db_path: str | Path) -> sqlite3.Connection: + # Cold-start race: GUI può avviarsi prima che il paper writer crei il file. + db_path_str = str(db_path) + deadline = time.monotonic() + 5.0 + while True: + try: + conn = sqlite3.connect(db_path_str, timeout=5.0) + conn.row_factory = sqlite3.Row + return conn + except sqlite3.OperationalError: + if time.monotonic() >= deadline: + raise + time.sleep(1.0) + + +def paper_runs_df(db_path: str | Path) -> pd.DataFrame: + with _paper_conn(db_path) as conn: + rows = conn.execute( + "SELECT id, name, started_at, stopped_at, status, initial_capital, config_json " + "FROM paper_trading_runs ORDER BY started_at DESC" + ).fetchall() + return pd.DataFrame([dict(r) for r in rows]) + + +def paper_equity_df(db_path: str | Path, run_id: str) -> pd.DataFrame: + with _paper_conn(db_path) as conn: + rows = conn.execute( + "SELECT ts, equity, cash, positions_value FROM paper_trading_equity " + "WHERE paper_run_id=? ORDER BY ts ASC", + (run_id,), + ).fetchall() + return pd.DataFrame([dict(r) for r in rows]) + + +def paper_positions_df(db_path: str | Path, run_id: str) -> pd.DataFrame: + with _paper_conn(db_path) as conn: + rows = conn.execute( + "SELECT symbol, side, qty, entry_price, entry_ts " + "FROM paper_trading_positions WHERE paper_run_id=? ORDER BY symbol", + (run_id,), + ).fetchall() + return pd.DataFrame([dict(r) for r in rows]) + + +def paper_trades_df(db_path: str | Path, run_id: str, limit: int = 100) -> pd.DataFrame: + with _paper_conn(db_path) as conn: + rows = conn.execute( + "SELECT symbol, side, qty, entry_price, exit_price, entry_ts, exit_ts, pnl, fees " + "FROM paper_trading_trades WHERE paper_run_id=? ORDER BY exit_ts DESC LIMIT ?", + (run_id, limit), + ).fetchall() + return pd.DataFrame([dict(r) for r in rows]) + + +def paper_ticks_df(db_path: str | Path, run_id: str, limit: int = 50) -> pd.DataFrame: + with _paper_conn(db_path) as conn: + rows = conn.execute( + "SELECT ts, bar_ts, symbol, close_price, signal, action_taken " + "FROM paper_trading_ticks WHERE paper_run_id=? ORDER BY ts DESC LIMIT ?", + (run_id, limit), + ).fetchall() + return pd.DataFrame([dict(r) for r in rows]) + + +def paper_run_summary(db_path: str | Path, run_id: str) -> dict[str, Any]: + """Aggrega metriche sintetiche per la pagina paper trading.""" + with _paper_conn(db_path) as conn: + run = conn.execute( + "SELECT id, name, started_at, stopped_at, status, initial_capital, config_json " + "FROM paper_trading_runs WHERE id=?", + (run_id,), + ).fetchone() + if run is None: + return {} + run = dict(run) + + eq_row = conn.execute( + "SELECT equity, cash, positions_value, ts FROM paper_trading_equity " + "WHERE paper_run_id=? ORDER BY ts DESC LIMIT 1", + (run_id,), + ).fetchone() + + trades_agg = conn.execute( + "SELECT COUNT(*) AS n, COALESCE(SUM(pnl),0) AS sum_pnl, " + "COALESCE(SUM(fees),0) AS sum_fees FROM paper_trading_trades " + "WHERE paper_run_id=?", + (run_id,), + ).fetchone() + + tick_agg = conn.execute( + "SELECT COUNT(*) AS n, MAX(ts) AS last_ts FROM paper_trading_ticks " + "WHERE paper_run_id=?", + (run_id,), + ).fetchone() + + positions_n = conn.execute( + "SELECT COUNT(*) AS n FROM paper_trading_positions WHERE paper_run_id=?", + (run_id,), + ).fetchone()["n"] + + initial = float(run["initial_capital"]) + current_equity = float(eq_row["equity"]) if eq_row is not None else initial + pnl_pct = (current_equity - initial) / initial * 100.0 if initial else 0.0 + + return { + "id": run["id"], + "name": run["name"], + "status": run["status"], + "started_at": run["started_at"], + "stopped_at": run["stopped_at"], + "initial_capital": initial, + "config": json.loads(run["config_json"]), + "current_equity": current_equity, + "current_cash": float(eq_row["cash"]) if eq_row is not None else initial, + "current_positions_value": float(eq_row["positions_value"]) if eq_row is not None else 0.0, + "last_equity_ts": eq_row["ts"] if eq_row is not None else None, + "pnl_abs": current_equity - initial, + "pnl_pct": pnl_pct, + "n_trades": int(trades_agg["n"]), + "trades_pnl": float(trades_agg["sum_pnl"]), + "trades_fees": float(trades_agg["sum_fees"]), + "n_ticks": int(tick_agg["n"]), + "last_tick_ts": tick_agg["last_ts"], + "n_open_positions": int(positions_n), + } + + +# --------------------------------------------------------------------------- +# Pythagoras-specific helpers (winners invariance + candle pattern usage) +# --------------------------------------------------------------------------- + + +def load_invariance_metrics(ga_db_path: str) -> "pd.DataFrame": + """Per ogni winner ritorna (genome_id, cognitive_style, fitness, sharpe_btc, sharpe_eth, invariance_score). + + Lo schema atteso e' la tabella ``pythagoras_winners`` creata dal runner + ``scripts/run_pythagoras_smoke.py`` (Task 6.1). + """ + import sqlite3 + + import pandas as pd + + con = sqlite3.connect(ga_db_path) + try: + return pd.read_sql_query( + "SELECT genome_id, cognitive_style, fitness, sharpe_btc, sharpe_eth, " + "invariance_score FROM pythagoras_winners ORDER BY fitness DESC", + con, + ) + finally: + con.close() + + +def load_candle_pattern_usage(ga_db_path: str) -> "pd.DataFrame": + """Per ogni winner estrae le sequenze candle_pattern usate (per heatmap).""" + import json + import sqlite3 + + import pandas as pd + + con = sqlite3.connect(ga_db_path) + try: + df = pd.read_sql_query( + "SELECT genome_id, cognitive_style, rules_json FROM pythagoras_winners", + con, + ) + finally: + con.close() + records: list[dict] = [] + for _, row in df.iterrows(): + rules = json.loads(row["rules_json"]).get("rules", []) + for r in rules: + for ind_name, params in _walk_indicators(r["condition"]): + if ind_name == "candle_pattern": + length = int(params[0]) + syms = [int(s) for s in params[1: 1 + length]] + seq_str = "".join({0: "U", 1: "D", 2: "0"}[s] for s in syms) + records.append( + { + "genome_id": row["genome_id"], + "cognitive_style": row["cognitive_style"], + "sequence": seq_str, + "length": length, + } + ) + return pd.DataFrame.from_records(records) + + +def _walk_indicators(node: dict): + """Yields (indicator_name, params) for every IndicatorNode in the AST.""" + if "op" in node: + for a in node.get("args", []): + yield from _walk_indicators(a) + elif node.get("kind") == "indicator": + yield node["name"], node["params"]