feat(strategy_pythagoras): port frontend data layer + invariance/candle helpers
This commit is contained in:
@@ -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"]
|
||||||
Reference in New Issue
Block a user