0.0.8

apps/pair_selector/pair_selector.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
 import asyncio
+import os
+import sqlite3
 from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Sequence, Set, Tuple, Union
 
 from aiohttp import web
 import numpy as np
@@ -26,6 +28,12 @@ from cvttpy_trading.trading.mkt_data.md_summary import MdTradesAggregate, MdSumm
 from pairs_trading.apps.pair_selector.renderer import HtmlRenderer
 
 
+@dataclass
+class BacktestAggregate:
+    aggr_time_ns_: int
+    num_trades_: Optional[int]
+
+
 @dataclass
 class InstrumentQuality(NamedObject):
     instrument_: ExchangeInstrument
@@ -51,6 +59,9 @@ class PairStats(NamedObject):
 
     def as_dict(self) -> Dict[str, Any]:
         return {
+            "exchange_a": self.instrument_a_.exchange_id_,
+            "exchange_b": self.instrument_b_.exchange_id_,
+            "pair_name": self.pair_name_,
             "instrument_a": self.instrument_a_.instrument_id(),
             "instrument_b": self.instrument_b_.instrument_id(),
             "pvalue_eg": self.pvalue_eg_,
@@ -64,6 +75,28 @@ class PairStats(NamedObject):
         }
 
 
+def _extract_price_from_fields(
+    price_field: str,
+    inst: ExchangeInstrument,
+    open: Optional[float],
+    high: Optional[float],
+    low: Optional[float],
+    close: Optional[float],
+    vwap: Optional[float],
+) -> float:
+    field_map = {
+        "open": open,
+        "high": high,
+        "low": low,
+        "close": close,
+        "vwap": vwap,
+    }
+    raw = field_map.get(price_field, close)
+    if raw is None:
+        raw = 0.0
+    return inst.get_price(raw)
+
+
 class DataFetcher(NamedObject):
     sender_: RESTSender
     interval_sec_: int
@@ -103,6 +136,9 @@ class DataFetcher(NamedObject):
         ]
 
 
+AggregateLike = Union[MdTradesAggregate, BacktestAggregate]
+
+
 class QualityChecker(NamedObject):
     interval_sec_: int
 
@@ -110,7 +146,10 @@ class QualityChecker(NamedObject):
         self.interval_sec_ = interval_sec
 
     def evaluate(
-        self, inst: ExchangeInstrument, aggr: List[MdTradesAggregate]
+        self,
+        inst: ExchangeInstrument,
+        aggr: Sequence[AggregateLike],
+        now_ts: Optional[pd.Timestamp] = None,
     ) -> InstrumentQuality:
         if len(aggr) == 0:
             return InstrumentQuality(
@@ -124,7 +163,7 @@ class QualityChecker(NamedObject):
         aggr_sorted = sorted(aggr, key=lambda a: a.aggr_time_ns_)
 
         latest_ts = pd.to_datetime(aggr_sorted[-1].aggr_time_ns_, unit="ns", utc=True)
-        now_ts = pd.Timestamp.utcnow()
+        now_ts = now_ts or pd.Timestamp.utcnow()
         recency_cutoff = now_ts - pd.Timedelta(seconds=2 * self.interval_sec_)
         if latest_ts <= recency_cutoff:
             return InstrumentQuality(
@@ -145,7 +184,7 @@ class QualityChecker(NamedObject):
             reason_=reason,
         )
 
-    def _check_gaps(self, aggr: List[MdTradesAggregate]) -> Tuple[bool, str]:
+    def _check_gaps(self, aggr: Sequence[AggregateLike]) -> Tuple[bool, str]:
         NUM_TRADES_THRESHOLD = 50
         if len(aggr) < 2:
             return True, "ok"
@@ -169,11 +208,11 @@ class QualityChecker(NamedObject):
         return True, "ok"
 
     @staticmethod
-    def _approximate_num_trades(prev_nt: int, next_nt: int) -> float:
+    def _approximate_num_trades(prev_nt: Optional[int], next_nt: Optional[int]) -> float:
         if prev_nt is None and next_nt is None:
             return 0.0
         if prev_nt is None:
-            return float(next_nt)
+            return float(next_nt or 0)
         if next_nt is None:
             return float(prev_nt)
         return (prev_nt + next_nt) / 2.0
@@ -206,6 +245,7 @@ class PairAnalyzer(NamedObject):
                 merged = pd.merge(df_a, df_b, on="tstamp", how="inner").sort_values(
                     "tstamp"
                 )
+                # Log.info(f"{self.fname()}: analyzing {pair_name}")
                 stats = self._compute_stats(inst_a, inst_b, pair_name, merged)
                 if stats:
                     results[pair_name] = stats
@@ -289,7 +329,7 @@ class PairAnalyzer(NamedObject):
         self._assign_ranks(ranked, key=lambda r: r.pvalue_adf_, attr="rank_adf_")
         self._assign_ranks(ranked, key=lambda r: r.pvalue_j_, attr="rank_j_")
         for res in ranked:
-            res.composite_rank_ = res.rank_eg_ + res.rank_adf_ + res.rank_j_
+            res.composite_rank_ = res.rank_eg_ + res.rank_adf_ # + res.rank_j_
         ranked.sort(key=lambda r: r.composite_rank_)
         return {res.pair_name_: res for res in ranked}
 
@@ -402,17 +442,15 @@ class PairSelectionEngine(NamedObject):
     def _extract_price(
         self, aggr: MdTradesAggregate, inst: ExchangeInstrument
     ) -> float:
-        price_field = self.price_field_
-        # MdTradesAggregate inherits hist bar with fields open_, high_, low_, close_, vwap_
-        field_map = {
-            "open": aggr.open_,
-            "high": aggr.high_,
-            "low": aggr.low_,
-            "close": aggr.close_,
-            "vwap": aggr.vwap_,
-        }
-        raw = field_map.get(price_field, aggr.close_)
-        return inst.get_price(raw)
+        return _extract_price_from_fields(
+            price_field=self.price_field_,
+            inst=inst,
+            open=aggr.open_,
+            high=aggr.high_,
+            low=aggr.low_,
+            close=aggr.close_,
+            vwap=aggr.vwap_,
+        )
 
     def sleep_seconds_until_next_cycle(self) -> float:
         now_ns = current_nanoseconds()
@@ -443,13 +481,356 @@ class PairSelectionEngine(NamedObject):
         }
 
 
+class PairSelectionBacktest(NamedObject):
+    config_: object
+    instruments_: List[ExchangeInstrument]
+    price_field_: str
+    input_db_: str
+    output_db_: str
+    interval_sec_: int
+    history_depth_hours_: int
+    quality_: QualityChecker
+    analyzer_: PairAnalyzer
+    inst_by_key_: Dict[Tuple[str, str], ExchangeInstrument]
+    inst_by_id_: Dict[str, Optional[ExchangeInstrument]]
+    ambiguous_ids_: Set[str]
+
+    def __init__(
+        self,
+        config: Config,
+        instruments: List[ExchangeInstrument],
+        price_field: str,
+        input_db: str,
+        output_db: str,
+    ) -> None:
+        self.config_ = config
+        self.instruments_ = instruments
+        self.price_field_ = price_field
+        self.input_db_ = input_db
+        self.output_db_ = output_db
+
+        interval_sec = int(config.get_value("interval_sec", 0))
+        if interval_sec <= 0:
+            Log.warning(
+                f"{self.fname()}: interval_sec not set; defaulting to 60 seconds"
+            )
+            interval_sec = 60
+        history_depth_hours = int(config.get_value("history_depth_hours", 0))
+        assert history_depth_hours > 0, "history_depth_hours must be > 0"
+
+        self.interval_sec_ = interval_sec
+        self.history_depth_hours_ = history_depth_hours
+        self.quality_ = QualityChecker(interval_sec=interval_sec)
+        self.analyzer_ = PairAnalyzer(
+            price_field=price_field, interval_sec=interval_sec
+        )
+
+        self.inst_by_key_ = {
+            (inst.exchange_id_, inst.instrument_id()): inst for inst in instruments
+        }
+        self.inst_by_id_ = {}
+        self.ambiguous_ids_ = set()
+        for inst in instruments:
+            inst_id = inst.instrument_id()
+            if inst_id in self.inst_by_id_:
+                existing = self.inst_by_id_[inst_id]
+                if existing is not None and existing.exchange_id_ != inst.exchange_id_:
+                    self.inst_by_id_[inst_id] = None
+                    self.ambiguous_ids_.add(inst_id)
+            elif inst_id not in self.ambiguous_ids_:
+                self.inst_by_id_[inst_id] = inst
+
+        if self.ambiguous_ids_:
+            Log.warning(
+                f"{self.fname()}: ambiguous instrument_id(s) without exchange_id: "
+                f"{sorted(self.ambiguous_ids_)}"
+            )
+
+    def run(self) -> None:
+        df = self._load_input_df()
+        if df.empty:
+            Log.warning(f"{self.fname()}: no rows in md_1min_bars")
+            return
+
+        df = self._filter_instruments(df)
+        if df.empty:
+            Log.warning(f"{self.fname()}: no rows after instrument filtering")
+            return
+
+        conn = self._init_output_db()
+        try:
+            self._run_backtest(df, conn)
+        finally:
+            conn.commit()
+            conn.close()
+
+    def _load_input_df(self) -> pd.DataFrame:
+        if not os.path.exists(self.input_db_):
+            raise FileNotFoundError(f"input_db not found: {self.input_db_}")
+        with sqlite3.connect(self.input_db_) as conn:
+            df = pd.read_sql_query(
+                """
+                SELECT
+                    tstamp,
+                    tstamp_ns,
+                    exchange_id,
+                    instrument_id,
+                    open,
+                    high,
+                    low,
+                    close,
+                    volume,
+                    vwap,
+                    num_trades
+                FROM md_1min_bars
+                """,
+                conn,
+            )
+        if df.empty:
+            return df
+
+        ts_ns = pd.to_datetime(df["tstamp_ns"], unit="ns", utc=True, errors="coerce")
+        ts_txt = pd.to_datetime(df["tstamp"], utc=True, errors="coerce")
+        df["tstamp"] = ts_ns.fillna(ts_txt)
+        df = df.dropna(subset=["tstamp", "instrument_id"]).copy()
+        df["exchange_id"] = df["exchange_id"].fillna("")
+        df["instrument_id"] = df["instrument_id"].astype(str)
+        df["tstamp_ns"] = df["tstamp"].astype("int64")
+        return df.sort_values("tstamp").reset_index(drop=True)
+
+    def _filter_instruments(self, df: pd.DataFrame) -> pd.DataFrame:
+        instrument_ids = {inst.instrument_id() for inst in self.instruments_}
+        df = df[df["instrument_id"].isin(instrument_ids)].copy()
+        if "exchange_id" in df.columns:
+            exchange_ids = {inst.exchange_id_ for inst in self.instruments_}
+            df = df[
+                (df["exchange_id"].isin(exchange_ids)) | (df["exchange_id"] == "")
+            ].copy()
+        return df
+
+    def _init_output_db(self) -> sqlite3.Connection:
+        if os.path.exists(self.output_db_):
+            os.remove(self.output_db_)
+        conn = sqlite3.connect(self.output_db_)
+        conn.execute(
+            """
+            CREATE TABLE pair_selection_history (
+                tstamp TEXT,
+                tstamp_ns INTEGER,
+                pair_name TEXT,
+                exchange_a TEXT,
+                instrument_a TEXT,
+                exchange_b TEXT,
+                instrument_b TEXT,
+                pvalue_eg REAL,
+                pvalue_adf REAL,
+                pvalue_j REAL,
+                trace_stat_j REAL,
+                rank_eg INTEGER,
+                rank_adf INTEGER,
+                rank_j INTEGER,
+                composite_rank REAL
+            )
+            """
+        )
+        conn.execute(
+            """
+            CREATE INDEX idx_pair_selection_history_pair_name
+            ON pair_selection_history (pair_name)
+            """
+        )
+        conn.execute(
+            """
+            CREATE UNIQUE INDEX idx_pair_selection_history_tstamp_pair
+            ON pair_selection_history (tstamp, pair_name)
+            """
+        )
+        conn.commit()
+        return conn
+
+    def _resolve_instrument(
+        self, exchange_id: str, instrument_id: str
+    ) -> Optional[ExchangeInstrument]:
+        if exchange_id:
+            inst = self.inst_by_key_.get((exchange_id, instrument_id))
+            if inst is not None:
+                return inst
+        inst = self.inst_by_id_.get(instrument_id)
+        if inst is None and instrument_id in self.ambiguous_ids_:
+            return None
+        return inst
+
+    def _build_day_series(
+        self, df_day: pd.DataFrame
+    ) -> Dict[ExchangeInstrument, pd.DataFrame]:
+        series: Dict[ExchangeInstrument, pd.DataFrame] = {}
+        group_cols = ["exchange_id", "instrument_id"]
+        for key, group in df_day.groupby(group_cols, dropna=False):
+            exchange_id, instrument_id = key
+            inst = self._resolve_instrument(str(exchange_id or ""), str(instrument_id))
+            if inst is None:
+                continue
+            df_inst = group.copy()
+            df_inst["price"] = [
+                _extract_price_from_fields(
+                    price_field=self.price_field_,
+                    inst=inst,
+                    open=float(row.open), #type: ignore
+                    high=float(row.high), #type: ignore
+                    low=float(row.low), #type: ignore
+                    close=float(row.close), #type: ignore
+                    vwap=float(row.vwap),#type: ignore
+                )
+                for row in df_inst.itertuples(index=False)
+            ]
+            df_inst = df_inst[["tstamp", "tstamp_ns", "price", "num_trades"]]
+            if inst in series:
+                series[inst] = pd.concat([series[inst], df_inst], ignore_index=True)
+            else:
+                series[inst] = df_inst
+        for inst in list(series.keys()):
+            series[inst] = series[inst].sort_values("tstamp").reset_index(drop=True)
+        return series
+
+    def _run_backtest(self, df: pd.DataFrame, conn: sqlite3.Connection) -> None:
+        window_minutes = self.history_depth_hours_ * 60
+        window_td = pd.Timedelta(minutes=window_minutes)
+        step_td = pd.Timedelta(seconds=self.interval_sec_)
+
+        df = df.copy()
+        df["day"] = df["tstamp"].dt.normalize()
+        days = sorted(df["day"].unique())
+        for day in days:
+            day_label = pd.Timestamp(day).date()
+            df_day = df[df["day"] == day]
+            t0 = df_day["tstamp"].min()
+            t_last = df_day["tstamp"].max()
+            if t_last - t0 < window_td:
+                Log.warning(
+                    f"{self.fname()}: skipping {day_label} (insufficient data)"
+                )
+                continue
+
+            day_series = self._build_day_series(df_day)
+            if len(day_series) < 2:
+                Log.warning(
+                    f"{self.fname()}: skipping {day_label} (insufficient instruments)"
+                )
+                continue
+
+            start = t0
+            expected_end = start + window_td
+            while expected_end <= t_last:
+                window_slices: Dict[ExchangeInstrument, pd.DataFrame] = {}
+                ts: Optional[pd.Timestamp] = None
+                for inst, df_inst in day_series.items():
+                    df_win = df_inst[
+                        (df_inst["tstamp"] >= start)
+                        & (df_inst["tstamp"] < expected_end)
+                    ]
+                    if df_win.empty:
+                        continue
+                    window_slices[inst] = df_win
+                    last_ts = df_win["tstamp"].iloc[-1]
+                    if ts is None or last_ts > ts:
+                        ts = last_ts
+
+                if window_slices and ts is not None:
+                    price_series: Dict[ExchangeInstrument, pd.DataFrame] = {}
+                    for inst, df_win in window_slices.items():
+                        aggr = self._to_backtest_aggregates(df_win)
+                        q = self.quality_.evaluate(
+                            inst=inst, aggr=aggr, now_ts=ts
+                        )
+                        if q.status_ != "PASS":
+                            continue
+                        price_series[inst] = df_win[["tstamp", "price"]]
+                    pair_results = self.analyzer_.analyze(price_series)
+                    Log.info(f"{self.fname()}: Saving Results for window ending {ts}")
+                    self._insert_results(conn, ts, pair_results)
+
+                start = start + step_td
+                expected_end = start + window_td
+
+    @staticmethod
+    def _to_backtest_aggregates(df_win: pd.DataFrame) -> List[BacktestAggregate]:
+        aggr: List[BacktestAggregate] = []
+        for tstamp_ns, num_trades in zip(df_win["tstamp_ns"], df_win["num_trades"]):
+            nt = None if pd.isna(num_trades) else int(num_trades)
+            aggr.append(
+                BacktestAggregate(aggr_time_ns_=int(tstamp_ns), num_trades_=nt)
+            )
+        return aggr
+
+    @staticmethod
+    def _insert_results(
+        conn: sqlite3.Connection,
+        ts: pd.Timestamp,
+        pair_results: Dict[str, PairStats],
+    ) -> None:
+        if not pair_results:
+            return
+        iso = ts.isoformat()
+        ns = int(ts.value)
+        rows = []
+        for pair_name in sorted(pair_results.keys()):
+            stats = pair_results[pair_name]
+            rows.append(
+                (
+                    iso,
+                    ns,
+                    pair_name,
+                    stats.instrument_a_.exchange_id_,
+                    stats.instrument_a_.instrument_id(),
+                    stats.instrument_b_.exchange_id_,
+                    stats.instrument_b_.instrument_id(),
+                    stats.pvalue_eg_,
+                    stats.pvalue_adf_,
+                    stats.pvalue_j_,
+                    stats.trace_stat_j_,
+                    stats.rank_eg_,
+                    stats.rank_adf_,
+                    stats.rank_j_,
+                    stats.composite_rank_,
+                )
+            )
+        conn.executemany(
+            """
+            INSERT INTO pair_selection_history (
+                tstamp,
+                tstamp_ns,
+                pair_name,
+                exchange_a,
+                instrument_a,
+                exchange_b,
+                instrument_b,
+                pvalue_eg,
+                pvalue_adf,
+                pvalue_j,
+                trace_stat_j,
+                rank_eg,
+                rank_adf,
+                rank_j,
+                composite_rank
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            rows,
+        )
+        conn.commit()
+
+
+
 class PairSelector(NamedObject):
     instruments_: List[ExchangeInstrument]
     engine_: PairSelectionEngine
-    rest_service_: RestService
+    rest_service_: Optional[RestService]
+    backtest_: Optional[PairSelectionBacktest]
 
     def __init__(self) -> None:
         App.instance().add_cmdline_arg("--oneshot", action="store_true", default=False)
+        App.instance().add_cmdline_arg("--backtest", action="store_true", default=False)
+        App.instance().add_cmdline_arg("--input_db", default=None)
+        App.instance().add_cmdline_arg("--output_db", default=None)
         App.instance().add_call(App.Stage.Config, self._on_config())
         App.instance().add_call(App.Stage.Run, self.run())
 
@@ -458,6 +839,24 @@ class PairSelector(NamedObject):
         self.instruments_ = self._load_instruments(cfg)
         price_field = cfg.get_value("model/stat_model_price", "close")
 
+        self.backtest_ = None
+        self.rest_service_ = None
+        if App.instance().get_argument("backtest", False):
+            input_db = App.instance().get_argument("input_db", None)
+            output_db = App.instance().get_argument("output_db", None)
+            if not input_db or not output_db:
+                raise ValueError(
+                    "--input_db and --output_db are required when --backtest is set"
+                )
+            self.backtest_ = PairSelectionBacktest(
+                config=cfg,
+                instruments=self.instruments_,
+                price_field=price_field,
+                input_db=input_db,
+                output_db=output_db,
+            )
+            return
+
         self.engine_ = PairSelectionEngine(
             config=cfg,
             instruments=self.instruments_,
@@ -499,6 +898,11 @@ class PairSelector(NamedObject):
         return instruments
 
     async def run(self) -> None:
+        if App.instance().get_argument("backtest", False):
+            if self.backtest_ is None:
+                raise RuntimeError("backtest runner not initialized")
+            self.backtest_.run()
+            return
         oneshot = App.instance().get_argument("oneshot", False)
         while True:
             await self.engine_.run_once()