#!/usr/bin/env python3 """ strategy_registry.py SQLite-based persistent strategy registry for MQL5 CSV exports. Provides database initialization, schema migrations, CSV ingestion, watch-folder automation, and a query interface for cross-run lookups. Requires: Python 3.8+, pandas >= 1.5 (all other deps are stdlib). Usage: python strategy_registry.py """ import os import sqlite3 import hashlib import logging import warnings import glob import time import shutil from datetime import datetime, timezone from typing import Optional, List, Dict import pandas as pd warnings.filterwarnings("ignore", category=UserWarning) # ── Configuration ───────────────────────────────────────────────────────────── DEFAULT_DB_PATH = "strategy_registry.db" REGISTRY_VERSION = "v1_2_add_broker_fields" logging.basicConfig( filename="registry_audit.log", level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S" ) # ── Schema SQL ──────────────────────────────────────────────────────────────── SQL_CREATE_SCHEMA_VERSION = """ CREATE TABLE IF NOT EXISTS schema_version ( version_id INTEGER PRIMARY KEY, version_tag TEXT NOT NULL UNIQUE, applied_at TEXT NOT NULL, description TEXT ); """ SQL_CREATE_INGEST_RUNS = """ CREATE TABLE IF NOT EXISTS ingest_runs ( run_id INTEGER PRIMARY KEY AUTOINCREMENT, source_file TEXT NOT NULL, file_hash_sha1 TEXT NOT NULL, ingest_at_utc TEXT NOT NULL, row_count INTEGER NOT NULL DEFAULT 0, ea_version TEXT, notes TEXT ); """ SQL_CREATE_STRATEGY_RESULTS = """ CREATE TABLE IF NOT EXISTS strategy_results ( result_id INTEGER PRIMARY KEY AUTOINCREMENT, ingest_run_id INTEGER NOT NULL REFERENCES ingest_runs(run_id), test_phase TEXT, symbol TEXT, timeframe TEXT, indicator_name TEXT, filter_period INTEGER, param_sama_lr REAL, net_profit_usd REAL, sortino_ratio REAL, max_drawdown_pct REAL, profit_factor REAL, expected_payoff REAL, win_rate_pct REAL, total_trades INTEGER, sharpe_ratio REAL, recovery_factor REAL, false_flips_whipsaws INTEGER, avg_lag_on_turn_bars REAL, avg_dist_points REAL, intersection_freq_pct REAL, avg_trade_duration_hrs REAL, market_share_time_pct REAL, stability_over_years INTEGER, broker_server TEXT, account_currency TEXT, commission_bps REAL, avg_slippage_points REAL, run_timestamp_utc TEXT, ea_version TEXT, run_unique_id TEXT ); """ SQL_CREATE_STRATEGY_CONFIGS = """ CREATE TABLE IF NOT EXISTS strategy_configs ( config_id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, timeframe TEXT NOT NULL, indicator_name TEXT NOT NULL, filter_period INTEGER NOT NULL, param_sama_lr REAL DEFAULT 0.0, first_seen_utc TEXT NOT NULL, last_seen_utc TEXT NOT NULL, total_runs INTEGER NOT NULL DEFAULT 1, UNIQUE(symbol, timeframe, indicator_name, filter_period, param_sama_lr) ); """ SQL_CREATE_INDEXES = [ """CREATE INDEX IF NOT EXISTS idx_results_core ON strategy_results(symbol, test_phase, indicator_name);""", """CREATE INDEX IF NOT EXISTS idx_results_params ON strategy_results(indicator_name, filter_period, param_sama_lr);""", """CREATE INDEX IF NOT EXISTS idx_results_sortino ON strategy_results(sortino_ratio DESC);""", """CREATE INDEX IF NOT EXISTS idx_results_symbol_tf ON strategy_results(symbol, timeframe);""", """CREATE INDEX IF NOT EXISTS idx_ingest_hash ON ingest_runs(file_hash_sha1);""", ] # ── Migration Definitions ───────────────────────────────────────────────────── MIGRATIONS = [ ( "v1_0_initial", "Initial schema: schema_version, ingest_runs, " "strategy_results, strategy_configs", [ SQL_CREATE_SCHEMA_VERSION, SQL_CREATE_INGEST_RUNS, SQL_CREATE_STRATEGY_RESULTS, SQL_CREATE_STRATEGY_CONFIGS, ] + SQL_CREATE_INDEXES ), ( "v1_1_add_run_uid_index", "Add index on run_unique_id for fast deduplication lookups", [ """CREATE INDEX IF NOT EXISTS idx_results_run_uid ON strategy_results(run_unique_id);""" ] ), ( "v1_2_add_broker_fields", "Ensure broker_server and commission_bps columns exist", [ "ALTER TABLE strategy_results ADD COLUMN broker_server TEXT;", "ALTER TABLE strategy_results ADD COLUMN commission_bps REAL;", ] ), ] # ── Column name mapping: CSV header -> strategy_results column ──────────────── COLUMN_MAP = { "test_phase" : "test_phase", "symbol" : "symbol", "timeframe" : "timeframe", "indicator_name" : "indicator_name", "filter_period" : "filter_period", "param_sama_lr" : "param_sama_lr", "net_profit_$" : "net_profit_usd", "net_profit_usd" : "net_profit_usd", "net_profit" : "net_profit_usd", "sortino_ratio" : "sortino_ratio", "max_drawdown_pct" : "max_drawdown_pct", "max_drawdown_%" : "max_drawdown_pct", "profit_factor" : "profit_factor", "expected_payoff" : "expected_payoff", "win_rate_%" : "win_rate_pct", "win_rate_pct" : "win_rate_pct", "total_trades" : "total_trades", "sharpe_ratio" : "sharpe_ratio", "recovery_factor" : "recovery_factor", "false_flips_whipsaws" : "false_flips_whipsaws", "avg_lag_on_turn_bars" : "avg_lag_on_turn_bars", "avg_price_to_line_dist_pts" : "avg_dist_points", "avg_dist_points" : "avg_dist_points", "intersection_freq_%" : "intersection_freq_pct", "avg_trade_duration_hrs" : "avg_trade_duration_hrs", "market_share_time_%" : "market_share_time_pct", "stability_over_years" : "stability_over_years", "broker_server" : "broker_server", "account_currency" : "account_currency", "commission_bps" : "commission_bps", "avg_slippage_points" : "avg_slippage_points", "run_timestamp_utc" : "run_timestamp_utc", "ea_version" : "ea_version", "run_unique_id" : "run_unique_id", } RESULT_COLUMNS = [ "ingest_run_id", "test_phase", "symbol", "timeframe", "indicator_name", "filter_period", "param_sama_lr", "net_profit_usd", "sortino_ratio", "max_drawdown_pct", "profit_factor", "expected_payoff", "win_rate_pct", "total_trades", "sharpe_ratio", "recovery_factor", "false_flips_whipsaws", "avg_lag_on_turn_bars", "avg_dist_points", "intersection_freq_pct", "avg_trade_duration_hrs", "market_share_time_pct", "stability_over_years", "broker_server", "account_currency", "commission_bps", "avg_slippage_points", "run_timestamp_utc", "ea_version", "run_unique_id", ] # ───────────────────────────────────────────────────────────────────────────── # Connection and Migration # ───────────────────────────────────────────────────────────────────────────── def get_connection(db_path: str = DEFAULT_DB_PATH) -> sqlite3.Connection: conn = sqlite3.connect(db_path, detect_types=sqlite3.PARSE_DECLTYPES) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode = WAL;") conn.execute("PRAGMA foreign_keys = ON;") conn.execute("PRAGMA synchronous = NORMAL;") return conn def _applied_migrations(conn: sqlite3.Connection) -> set: try: rows = conn.execute( "SELECT version_tag FROM schema_version;" ).fetchall() return {r["version_tag"] for r in rows} except sqlite3.OperationalError: return set() def apply_migrations(conn: sqlite3.Connection): applied = _applied_migrations(conn) for version_tag, description, sql_list in MIGRATIONS: if version_tag in applied: continue print(f" [Migration] Applying: {version_tag}") with conn: for sql in sql_list: try: conn.execute(sql) except sqlite3.OperationalError as exc: if "duplicate column" in str(exc).lower(): continue raise conn.execute( "INSERT INTO schema_version " "(version_tag, applied_at, description) VALUES (?,?,?);", (version_tag, datetime.now(timezone.utc).isoformat(), description) ) logging.info(f"Migration applied: {version_tag}") print(f" [Migration] Applied: {version_tag}") def initialize_registry(db_path: str = DEFAULT_DB_PATH) -> sqlite3.Connection: is_new = not os.path.isfile(db_path) conn = get_connection(db_path) apply_migrations(conn) if is_new: print(f"[Registry] New database created: {db_path}") logging.info(f"New registry created: {db_path}") else: print(f"[Registry] Opened: {db_path}") return conn # ───────────────────────────────────────────────────────────────────────────── # CSV Ingestion # ───────────────────────────────────────────────────────────────────────────── def _sha1_of_file(file_path: str) -> str: h = hashlib.sha1() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(65536), b""): h.update(chunk) return h.hexdigest() def ingest_csv(conn : sqlite3.Connection, file_path : str, ea_version : str = "", notes : str = "", skip_dupes : bool = True) -> int: """ Ingests a CSV export file into strategy_results. Returns the number of rows inserted, or 0 if skipped as duplicate. """ if not os.path.isfile(file_path): print(f"[Ingestor] File not found: {file_path}") return 0 file_hash = _sha1_of_file(file_path) if skip_dupes: existing = conn.execute( "SELECT run_id FROM ingest_runs WHERE file_hash_sha1 = ?;", (file_hash,) ).fetchone() if existing: print(f"[Ingestor] Skipped (duplicate): {os.path.basename(file_path)}") logging.info(f"Duplicate skip: {file_path} (hash={file_hash[:8]})") return 0 for enc in ("cp1252", "utf-8", "latin-1"): try: df = pd.read_csv(file_path, encoding=enc, low_memory=False) break except (UnicodeDecodeError, LookupError): continue else: print(f"[Ingestor] Could not decode: {file_path}") return 0 if df.empty: print(f"[Ingestor] Empty file: {file_path}") return 0 df.columns = [c.strip().lower().replace(" ", "_") for c in df.columns] rename_map = {k: v for k, v in COLUMN_MAP.items() if k in df.columns} df = df.rename(columns=rename_map) now_utc = datetime.now(timezone.utc).isoformat() with conn: cursor = conn.execute( "INSERT INTO ingest_runs " "(source_file, file_hash_sha1, ingest_at_utc, " " row_count, ea_version, notes) " "VALUES (?,?,?,?,?,?);", (os.path.abspath(file_path), file_hash, now_utc, len(df), ea_version, notes) ) run_id = cursor.lastrowid insert_rows = [] for _, row in df.iterrows(): record = {"ingest_run_id": run_id} for col in RESULT_COLUMNS: if col == "ingest_run_id": continue val = row.get(col, None) try: if pd.isna(val): val = None except (TypeError, ValueError): pass record[col] = val insert_rows.append(record) if not insert_rows: return 0 placeholders = ", ".join(["?" for _ in RESULT_COLUMNS]) col_str = ", ".join(RESULT_COLUMNS) sql_insert = (f"INSERT INTO strategy_results ({col_str}) " f"VALUES ({placeholders});") with conn: conn.executemany( sql_insert, [[r.get(c) for c in RESULT_COLUMNS] for r in insert_rows] ) conn.execute( "UPDATE ingest_runs SET row_count = ? WHERE run_id = ?;", (len(insert_rows), run_id) ) _upsert_strategy_configs(conn, insert_rows, now_utc) logging.info(f"Ingested: {os.path.basename(file_path)} " f"| {len(insert_rows)} rows | run_id={run_id}") print(f"[Ingestor] {os.path.basename(file_path)}: " f"{len(insert_rows)} rows ingested (run_id={run_id}).") return len(insert_rows) def _upsert_strategy_configs(conn : sqlite3.Connection, rows : List[Dict], now_utc_str : str): seen_configs = set() with conn: for record in rows: sym = record.get("symbol") tf = record.get("timeframe") ind = record.get("indicator_name") per = record.get("filter_period") lr = record.get("param_sama_lr") or 0.0 if None in (sym, tf, ind, per): continue key = (sym, tf, ind, per, lr) if key in seen_configs: continue seen_configs.add(key) conn.execute( """INSERT OR IGNORE INTO strategy_configs (symbol, timeframe, indicator_name, filter_period, param_sama_lr, first_seen_utc, last_seen_utc, total_runs) VALUES (?,?,?,?,?,?,?,1);""", (sym, tf, ind, per, lr, now_utc_str, now_utc_str) ) conn.execute( """UPDATE strategy_configs SET last_seen_utc = ?, total_runs = total_runs + 1 WHERE symbol = ? AND timeframe = ? AND indicator_name = ? AND filter_period = ? AND param_sama_lr = ?;""", (now_utc_str, sym, tf, ind, per, lr) ) # ───────────────────────────────────────────────────────────────────────────── # Watch-Folder Auto-Ingestor # ───────────────────────────────────────────────────────────────────────────── def watch_folder(conn : sqlite3.Connection, watch_dir : str, pattern : str = "*.csv", poll_interval : int = 30, ea_version : str = "") -> None: """ Monitors a directory for new CSV files and ingests them automatically. Press Ctrl+C to stop. """ print(f"[WatchFolder] Monitoring: {watch_dir}") print(f"[WatchFolder] Pattern: {pattern} | Poll: {poll_interval}s") print(f"[WatchFolder] Press Ctrl+C to stop.\n") total_ingested = 0 while True: try: candidates = sorted(glob.glob(os.path.join(watch_dir, pattern))) for file_path in candidates: rows = ingest_csv(conn, file_path, ea_version=ea_version, notes="auto-ingested by watch_folder", skip_dupes=True) total_ingested += rows time.sleep(poll_interval) except KeyboardInterrupt: print(f"\n[WatchFolder] Stopped. " f"Total rows ingested: {total_ingested:,}") break except Exception as exc: print(f"[WatchFolder] Error: {exc}") logging.error(f"WatchFolder error: {exc}") time.sleep(poll_interval) # ───────────────────────────────────────────────────────────────────────────── # Query Interface # ───────────────────────────────────────────────────────────────────────────── def query(conn : sqlite3.Connection, sql : str, params: tuple = ()) -> pd.DataFrame: """Executes arbitrary SQL and returns a pandas DataFrame.""" return pd.read_sql_query(sql, conn, params=params) def get_all_results(conn : sqlite3.Connection, symbol : str = None, timeframe : str = None, test_phase : str = None, indicator : str = None, min_sortino : float = None, min_trades : int = None) -> pd.DataFrame: """Returns strategy_results rows matching the supplied filters.""" conditions = ["1=1"] params = [] if symbol: conditions.append("symbol = ?") params.append(symbol) if timeframe: conditions.append("timeframe = ?") params.append(timeframe) if test_phase: conditions.append("test_phase = ?") params.append(test_phase) if indicator: conditions.append("indicator_name = ?") params.append(indicator) if min_sortino is not None: conditions.append("sortino_ratio >= ?") params.append(min_sortino) if min_trades is not None: conditions.append("total_trades >= ?") params.append(min_trades) where = " AND ".join(conditions) sql = (f"SELECT * FROM strategy_results WHERE {where} " f"ORDER BY sortino_ratio DESC;") return query(conn, sql, tuple(params)) def get_best_per_indicator(conn : sqlite3.Connection, symbol : str, timeframe : str, test_phase : str = "InSample") -> pd.DataFrame: """Returns the single best result (by Sortino) per indicator.""" sql = """ SELECT r.* FROM strategy_results r INNER JOIN ( SELECT indicator_name, MAX(sortino_ratio) AS max_sortino FROM strategy_results WHERE symbol = ? AND timeframe = ? AND test_phase = ? AND sortino_ratio IS NOT NULL GROUP BY indicator_name ) best ON r.indicator_name = best.indicator_name AND r.sortino_ratio = best.max_sortino AND r.symbol = ? AND r.timeframe = ? AND r.test_phase = ? ORDER BY r.sortino_ratio DESC; """ return query(conn, sql, (symbol, timeframe, test_phase, symbol, timeframe, test_phase)) def get_walk_forward_pair(conn : sqlite3.Connection, symbol : str, timeframe : str, indicator : str, period : int, lr : float = 0.0) -> pd.DataFrame: """Retrieves the InSample and OutSample rows for an exact parameter combo.""" sql = """ SELECT * FROM strategy_results WHERE symbol = ? AND timeframe = ? AND indicator_name = ? AND filter_period = ? AND (param_sama_lr = ? OR param_sama_lr IS NULL) AND test_phase IN ('InSample', 'OutSample') ORDER BY test_phase ASC, run_timestamp_utc DESC; """ return query(conn, sql, (symbol, timeframe, indicator, period, lr)) def get_stability_trend(conn : sqlite3.Connection, symbol : str, indicator: str, period : int) -> pd.DataFrame: """Returns the time-ordered performance history of a specific configuration.""" sql = """ SELECT r.run_timestamp_utc, r.ea_version, r.test_phase, r.sortino_ratio, r.net_profit_usd, r.max_drawdown_pct, r.false_flips_whipsaws, i.source_file, i.ingest_at_utc FROM strategy_results r JOIN ingest_runs i ON r.ingest_run_id = i.run_id WHERE r.symbol = ? AND r.indicator_name = ? AND r.filter_period = ? ORDER BY r.run_timestamp_utc ASC; """ return query(conn, sql, (symbol, indicator, period)) def get_config_summary(conn: sqlite3.Connection) -> pd.DataFrame: """Returns the strategy_configs table with run count statistics.""" sql = """ SELECT symbol, timeframe, indicator_name, filter_period, param_sama_lr, total_runs, first_seen_utc, last_seen_utc FROM strategy_configs ORDER BY total_runs DESC, last_seen_utc DESC; """ return query(conn, sql) def rank_by_robustness(conn : sqlite3.Connection, symbol : str, timeframe : str, min_runs : int = 2, min_sortino : float = 0.0) -> pd.DataFrame: """ Ranks configurations by mean Sortino and coefficient of variation across multiple InSample runs. """ sql = """ SELECT indicator_name, filter_period, param_sama_lr, COUNT(*) AS run_count, ROUND(AVG(sortino_ratio), 4) AS mean_sortino, ROUND(AVG(net_profit_usd), 2) AS mean_profit_usd, ROUND(AVG(max_drawdown_pct),2) AS mean_dd_pct, ROUND(AVG(false_flips_whipsaws), 1) AS mean_whipsaws, ROUND(AVG(avg_lag_on_turn_bars), 2) AS mean_lag_bars FROM strategy_results WHERE symbol = ? AND timeframe = ? AND test_phase = 'InSample' AND sortino_ratio IS NOT NULL GROUP BY indicator_name, filter_period, param_sama_lr HAVING COUNT(*) >= ? AND AVG(sortino_ratio) >= ? ORDER BY mean_sortino DESC; """ return query(conn, sql, (symbol, timeframe, min_runs, min_sortino)) def compare_ea_versions(conn : sqlite3.Connection, symbol : str, indicator: str) -> pd.DataFrame: """Compares mean performance across EA versions for a symbol/indicator.""" sql = """ SELECT r.ea_version, COUNT(*) AS run_count, ROUND(AVG(r.sortino_ratio), 4) AS mean_sortino, ROUND(AVG(r.net_profit_usd), 2) AS mean_profit_usd, ROUND(AVG(r.max_drawdown_pct),2) AS mean_dd_pct, MIN(r.run_timestamp_utc) AS first_run, MAX(r.run_timestamp_utc) AS last_run FROM strategy_results r WHERE r.symbol = ? AND r.indicator_name = ? AND r.ea_version IS NOT NULL AND r.ea_version != '' GROUP BY r.ea_version ORDER BY MIN(r.run_timestamp_utc) ASC; """ return query(conn, sql, (symbol, indicator)) # ───────────────────────────────────────────────────────────────────────────── # Registry Maintenance # ───────────────────────────────────────────────────────────────────────────── def find_duplicate_rows(conn: sqlite3.Connection) -> pd.DataFrame: """Identifies rows sharing identical key values across columns.""" sql = """ SELECT symbol, timeframe, indicator_name, filter_period, param_sama_lr, test_phase, run_timestamp_utc, COUNT(*) AS duplicate_count FROM strategy_results GROUP BY symbol, timeframe, indicator_name, filter_period, param_sama_lr, test_phase, run_timestamp_utc HAVING COUNT(*) > 1 ORDER BY duplicate_count DESC; """ return query(conn, sql) def prune_old_runs(conn : sqlite3.Connection, keep_days : int = 90, dry_run : bool = True) -> int: """ Removes ingest_runs and their strategy_results older than keep_days days. Defaults to dry_run=True so no data is deleted without explicit confirmation. """ from datetime import timedelta cutoff = (datetime.now(timezone.utc) - timedelta(days=keep_days)).isoformat() old_runs = query( conn, "SELECT run_id, source_file, row_count, ingest_at_utc " "FROM ingest_runs WHERE ingest_at_utc < ?;", (cutoff,) ) if old_runs.empty: print(f"[Prune] No runs older than {keep_days} days found.") return 0 total_rows = int(old_runs["row_count"].sum()) print(f"[Prune] Found {len(old_runs)} runs ({total_rows:,} rows) " f"older than {keep_days} days.") if dry_run: print(f"[Prune] Dry-run mode: no rows deleted. " f"Pass dry_run=False to execute.") return total_rows run_ids = tuple(old_runs["run_id"].tolist()) placeholders = ",".join(["?" for _ in run_ids]) with conn: conn.execute( f"DELETE FROM strategy_results " f"WHERE ingest_run_id IN ({placeholders});", run_ids ) conn.execute( f"DELETE FROM ingest_runs WHERE run_id IN ({placeholders});", run_ids ) conn.execute("VACUUM;") logging.info(f"Pruned {len(old_runs)} runs / {total_rows} rows.") print(f"[Prune] Deleted {total_rows:,} rows from {len(old_runs)} runs.") return total_rows def backup_registry(db_path : str = DEFAULT_DB_PATH, backup_dir : str = "registry_backups") -> str: """Creates a timestamped copy of the registry database file.""" os.makedirs(backup_dir, exist_ok=True) ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") backup_name = f"strategy_registry_{ts}.db" backup_path = os.path.join(backup_dir, backup_name) shutil.copy2(db_path, backup_path) size_mb = os.path.getsize(backup_path) / (1024 * 1024) print(f"[Backup] Registry backed up: {backup_path} ({size_mb:.2f} MB)") logging.info(f"Backup created: {backup_path}") return backup_path # ───────────────────────────────────────────────────────────────────────────── # Entry Point # ───────────────────────────────────────────────────────────────────────────── if __name__ == "__main__": # ── Initialize or open the registry ────────────────────────────────────── conn = initialize_registry("strategy_registry.db") # ── Ingest sample CSV exports ───────────────────────────────────────────── for csv_file in sorted(glob.glob("exports/*.csv")): ingest_csv(conn, csv_file, ea_version="v1.0.0", notes="validation run") # ── Config summary ───────────────────────────────────────────────────────── config_df = get_config_summary(conn) if not config_df.empty: print("\n── Strategy Config Summary ───────────────────────────────────") print(config_df.to_string(index=False)) # ── All results ──────────────────────────────────────────────────────────── results_df = get_all_results(conn) if not results_df.empty: print(f"\n── Total Rows in Registry: {len(results_df):,} ─────────────") print(results_df[["symbol", "timeframe", "indicator_name", "filter_period", "sortino_ratio", "test_phase"]].head(10).to_string(index=False)) # ── Duplicate check ──────────────────────────────────────────────────────── dupes = find_duplicate_rows(conn) if not dupes.empty: print(f"\n── Duplicate Rows Found: {len(dupes)} ──────────────────────") print(dupes.to_string(index=False)) else: print("\n── No duplicate rows detected. ──────────────────────────────") # ── Backup before closing ────────────────────────────────────────────────── if os.path.isfile("strategy_registry.db"): backup_registry("strategy_registry.db", "registry_backups") conn.close() print("\n[Registry] Session complete.")