to discard

.gitignore

@@ -1,11 +1,10 @@
 # SpecStory explanation file
 __pycache__/
 __OLD__/
-.specstory/
 .history/
 .cursorindexingignore
 data
-.vscode/
+####.vscode/
 cvttpy
 # SpecStory explanation file
 .specstory/.what-is-this.md

README.md

@@ -43,7 +43,7 @@ Each configuration dictionary specifies:
 - `db_table_name`: The name of the table within the SQLite database.
 - `instruments`: A list of symbols to consider for forming trading pairs.
 - `trading_hours`: Defines the session start and end times, crucial for equity markets.
-- `price_column`: The column in the data to be used as the price (e.g., "close").
+- `stat_model_price`: The column in the data to be used as the price (e.g., "close").
 - `dis-equilibrium_open_trshld`: The threshold (in standard deviations) of the dis-equilibrium for opening a trade.
 - `dis-equilibrium_close_trshld`: The threshold (in standard deviations) of the dis-equilibrium for closing an open trade.
 - `training_minutes`: The length of the rolling window (in minutes) used to train the model (e.g., calculate cointegration, mean, and standard deviation of the dis-equilibrium).

configuration/crypto_zscore.cfg

@@ -1,33 +0,0 @@
-{
-    "security_type": "CRYPTO",
-    "data_directory": "./data/crypto",
-    "datafiles": [
-        "20250602.mktdata.ohlcv.db"
-    ],
-    "db_table_name": "md_1min_bars",
-    "exchange_id": "BNBSPOT",
-    "instrument_id_pfx": "PAIR-",
-    "funding_per_pair": 2000.0,
-
-    # ====== Trading Parameters ======
-    "price_column": "close",
-    "dis-equilibrium_open_trshld": 2.0,
-    "dis-equilibrium_close_trshld": 0.5,
-    "training_minutes": 120,
-    "fit_method_class": "pt_trading.z-score_rolling_fit.ZScoreRollingFit",
-
-    # ====== Stop Conditions ======
-    "stop_close_conditions": {
-        "profit": 2.0,
-        "loss": -0.5
-    }
-
-    # ====== End of Session Closeout ======
-    "close_outstanding_positions": true,
-    # "close_outstanding_positions": false,
-    "trading_hours": {
-        "begin_session": "9:30:00",
-        "end_session": "22:30:00",
-        "timezone": "America/New_York"
-    }
-}

configuration/equity.cfg

@@ -2,34 +2,26 @@
     "security_type": "EQUITY",
     "data_directory": "./data/equity",
     "datafiles": [
-        "202506*.mktdata.ohlcv.db",
+        "20250618.mktdata.ohlcv.db",
     ],
     "db_table_name": "md_1min_bars",
     "exchange_id": "ALPACA",
     "instrument_id_pfx": "STOCK-",
-    "exclude_instruments": ["CAN"],
-
-    "funding_per_pair": 2000.0,
-
-    # ====== Trading Parameters ======
+    "trading_hours": {
+        "begin_session": "9:30:00",
+        "end_session": "16:00:00",
+        "timezone": "America/New_York"
+    },
     "price_column": "close",
+    "min_required_points": 30,
+    "zero_threshold": 1e-10,
     "dis-equilibrium_open_trshld": 2.0,
     "dis-equilibrium_close_trshld": 1.0,
     "training_minutes": 120,
-    "fit_method_class": "pt_trading.vecm_rolling_fit.VECMRollingFit",
-
-    # ====== Stop Conditions ======
-    "stop_close_conditions": {
-        "profit": 2.0,
-        "loss": -0.5
-    }
-
-    # ====== End of Session Closeout ======
-    "close_outstanding_positions": true,
-    # "close_outstanding_positions": false,
-    "trading_hours": {
-        "begin_session": "9:30:00",
-        "end_session": "15:30:00",
-        "timezone": "America/New_York"
-    }
+    "funding_per_pair": 2000.0,
+    # "fit_method_class": "pt_trading.sliding_fit.SlidingFit",
+    "fit_method_class": "pt_trading.static_fit.StaticFit",
+    "exclude_instruments": ["CAN"],
+    "close_outstanding_positions": false
+    
 }

configuration/equity_lg.cfg

@@ -0,0 +1,26 @@
+{
+    "security_type": "EQUITY",
+    "data_directory": "./data/equity",
+    "datafiles": [
+        "20250602.mktdata.ohlcv.db",
+    ],
+    "db_table_name": "md_1min_bars",
+    "exchange_id": "ALPACA",
+    "instrument_id_pfx": "STOCK-",
+    "trading_hours": {
+        "begin_session": "9:30:00",
+        "end_session": "16:00:00",
+        "timezone": "America/New_York"
+    },
+    "price_column": "close",
+    "min_required_points": 30,
+    "zero_threshold": 1e-10,
+    "dis-equilibrium_open_trshld": 2.0,
+    "dis-equilibrium_close_trshld": 1.0,
+    "training_minutes": 120,
+    "funding_per_pair": 2000.0,
+    "fit_method_class": "pt_trading.fit_methods.StaticFit",
+    "exclude_instruments": ["CAN"]
+}
+# "fit_method_class": "pt_trading.fit_methods.SlidingFit",
+# "fit_method_class": "pt_trading.fit_methods.StaticFit",

configuration/equity_zscore.cfg

@@ -1,35 +0,0 @@
-{
-    "security_type": "EQUITY",
-    "data_directory": "./data/equity",
-    "datafiles": [
-        "202506*.mktdata.ohlcv.db",
-    ],
-    "db_table_name": "md_1min_bars",
-    "exchange_id": "ALPACA",
-    "instrument_id_pfx": "STOCK-",
-    "exclude_instruments": ["CAN"],
-
-    "funding_per_pair": 2000.0,
-
-    # ====== Trading Parameters ======
-    "price_column": "close",
-    "dis-equilibrium_open_trshld": 2.0,
-    "dis-equilibrium_close_trshld": 1.0,
-    "training_minutes": 120,
-    "fit_method_class": "pt_trading.z-score_rolling_fit.ZScoreRollingFit",
-
-    # ====== Stop Conditions ======
-    "stop_close_conditions": {
-        "profit": 2.0,
-        "loss": -0.5
-    }
-
-    # ====== End of Session Closeout ======
-    "close_outstanding_positions": true,
-    # "close_outstanding_positions": false,
-    "trading_hours": {
-        "begin_session": "9:30:00",
-        "end_session": "15:30:00",
-        "timezone": "America/New_York"
-    }
-}

configuration/vecm.cfg

@@ -1,16 +1,26 @@
 {
-    "security_type": "CRYPTO",
+    "market_data_loading": {
+        "CRYPTO": {
     "data_directory": "./data/crypto",
-    "datafiles": [
-        "2025*.mktdata.ohlcv.db"
-    ],
     "db_table_name": "md_1min_bars",
-    "exchange_id": "BNBSPOT",
     "instrument_id_pfx": "PAIR-",
-    "funding_per_pair": 2000.0,
+        },
+        "EQUITY": {
+            "data_directory": "./data/equity",
+            "db_table_name": "md_1min_bars",
+            "instrument_id_pfx": "STOCK-",
+        }
+    },
 
+    # ====== Funding ======
+    "funding_per_pair": 2000.0,
+  
     # ====== Trading Parameters ======
-    "price_column": "close",
+    "stat_model_price": "close", # "vwap"
+    "execution_price": {
+        "column": "vwap",
+        "shift": 1,
+    },
     "dis-equilibrium_open_trshld": 2.0,
     "dis-equilibrium_close_trshld": 1.0,
     "training_minutes": 120,
@@ -26,8 +36,8 @@
     "close_outstanding_positions": true,
     # "close_outstanding_positions": false,
     "trading_hours": {
+        "timezone": "America/New_York",
         "begin_session": "9:30:00",
-        "end_session": "21:30:00",
-        "timezone": "America/New_York"
+        "end_session": "18:30:00",
     }
 }

configuration/zscore.cfg

@@ -13,9 +13,13 @@
     },
 
     # ====== Funding ======
-            "funding_per_pair": 2000.0,
+    "funding_per_pair": 2000.0,
     # ====== Trading Parameters ======
-    "price_column": "close",
+    "stat_model_price": "close",
+    "execution_price": {
+        "column": "vwap",
+        "shift": 1,
+    },
     "dis-equilibrium_open_trshld": 2.0,
     "dis-equilibrium_close_trshld": 0.5,
     "training_minutes": 120,
@@ -31,8 +35,8 @@
     "close_outstanding_positions": true,
     # "close_outstanding_positions": false,
     "trading_hours": {
+        "timezone": "America/New_York",
         "begin_session": "9:30:00",
-        "end_session": "22:30:00",
-        "timezone": "America/New_York"
+        "end_session": "18:30:00",
     }
 }

lg_notes.md

@@ -0,0 +1,115 @@
+07.11.2025
+pairs_trading/configuration <---- directory for config
+equity_lg.cfg <-------- copy of equity.cfg
+How to run a Program: TRIANGLEsquare ----> triangle EQUITY backtest
+Results are in  > results (timestamp table for all runs)
+table "...timestamp... .pt_backtest_results.equity.db"
+going to table  using sqlite
+> sqlite3 '/home/coder/results/20250721_175750.pt_backtest_results.equity.db'
+
+sqlite> .databases
+main: /home/coder/results/20250717_180122.pt_backtest_results.equity.db r/w
+sqlite> .tables
+config                 outstanding_positions  pt_bt_results   
+
+sqlite> PRAGMA table_info('pt_bt_results');
+0|date|DATE|0||0
+1|pair|TEXT|0||0
+2|symbol|TEXT|0||0
+3|open_time|DATETIME|0||0
+4|open_side|TEXT|0||0
+5|open_price|REAL|0||0
+6|open_quantity|INTEGER|0||0
+7|open_disequilibrium|REAL|0||0
+8|close_time|DATETIME|0||0
+9|close_side|TEXT|0||0
+10|close_price|REAL|0||0
+11|close_quantity|INTEGER|0||0
+12|close_disequilibrium|REAL|0||0
+13|symbol_return|REAL|0||0
+14|pair_return|REAL|0||0
+
+select count(*) as cnt from pt_bt_results;
+8
+
+select * from pt_bt_results;
+
+select 
+date, close_time, pair, symbol, symbol_return, pair_return
+from pt_bt_results ;
+
+select date, sum(symbol_return) as daily_return 
+from pt_bt_results where date = '2025-06-18' group by date;
+
+.quit
+
+sqlite3 '/home/coder/results/20250717_172435.pt_backtest_results.equity.db'
+
+sqlite> select date, sum(symbol_return) as daily_return 
+from pt_bt_results group by date;
+
+2025-06-02|1.29845390060828
+...
+2025-06-18|-43.5084977104115 <========== ????? ==========>
+2025-06-20|11.8605547517183
+
+
+select 
+date, close_time, pair, symbol, symbol_return, pair_return
+from pt_bt_results ;
+
+select date, close_time, pair, symbol, symbol_return, pair_return
+from pt_bt_results where date = '2025-06-18';
+
+
+./scripts/load_equity_pair_intraday.sh -A NVDA -B QQQ -d 20250701  -T ./intraday_md
+
+to inspect exactly what sources, formats, and processing steps you can open the script with:
+head -n 50 ./scripts/load_equity_pair_intraday.sh
+
+
+
+✓ Data file found: /home/coder/pairs_trading/data/crypto/20250605.mktdata.ohlcv.db
+
+sqlite3 '/home/coder/results/20250722_201930.pt_backtest_results.crypto.db'
+
+sqlite3 '/home/coder/results/xxxxxxxx_yyyyyy.pt_backtest_results.pseudo.db'
+
+11111111
+=== At your terminal, run these commands:
+sqlite3 '/home/coder/results/20250722_201930.pt_backtest_results.crypto.db'
+=== Then inside the SQLite prompt:
+.mode csv
+.headers on
+.output results_20250722.csv
+SELECT * FROM pt_bt_results;
+.output stdout
+.quit
+
+cd /home/coder/
+
+# === mode csv formats output as CSV
+# === headers on includes column names
+# === output my_table.csv directs output to that file
+# === Run your SELECT query, then revert output
+# === Open my_table.csv in Excel directly
+
+# ======== Using scp (Secure Copy)
+# === On your local machine, open a terminal and run:
+scp cvtt@953f6e8df266:/home/coder/results_20250722.csv ~/Downloads/
+
+
+# ===== convert cvs pandas dataframe ====== -->
+import pandas as pd
+# Replace with the actual path to your CSV file
+file_path = '/home/coder/results_20250722.csv'
+# Read the CSV file into a DataFrame
+df = pd.read_csv(file_path)
+# Show the first few rows
+print(df.head())
+
+
+
+
+
+

lib/pt_trading/fit_method.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
 from enum import Enum
 from typing import Dict, Optional, cast
@@ -21,6 +23,15 @@ class PairsTradingFitMethod(ABC):
         "signed_scaled_disequilibrium",
         "pair",
     ]
+    @staticmethod
+    def create(config: Dict) -> PairsTradingFitMethod:
+        import importlib
+        fit_method_class_name = config.get("fit_method_class", None)
+        assert fit_method_class_name is not None
+        module_name, class_name = fit_method_class_name.rsplit(".", 1)
+        module = importlib.import_module(module_name)
+        fit_method = getattr(module, class_name)()
+        return cast(PairsTradingFitMethod, fit_method)
 
     @abstractmethod
     def run_pair(
@@ -32,6 +43,10 @@ class PairsTradingFitMethod(ABC):
 
     @abstractmethod
     def create_trading_pair(
-        self, config: Dict, market_data: pd.DataFrame, symbol_a: str, symbol_b: str, price_column: str
+        self,
+        config: Dict,
+        market_data: pd.DataFrame,
+        symbol_a: str,
+        symbol_b: str,
     ) -> TradingPair: ...
 

lib/pt_trading/results.py

@@ -121,7 +121,7 @@ def store_config_in_database(
     config_file_path: str,
     config: Dict,
     fit_method_class: str,
-    datafiles: List[str],
+    datafiles: List[Tuple[str, str]],
     instruments: List[Dict[str, str]],
 ) -> None:
     """
@@ -140,7 +140,7 @@ def store_config_in_database(
         config_json = json.dumps(config, indent=2, default=str)
 
         # Convert lists to comma-separated strings for storage
-        datafiles_str = ", ".join(datafiles)
+        datafiles_str = ", ".join([f"{datafile}" for _, datafile in datafiles])
         instruments_str = ", ".join(
             [
                 f"{inst['symbol']}:{inst['instrument_type']}:{inst['exchange_id']}"
@@ -417,7 +417,7 @@ class BacktestResult:
 
             # Print pair returns with disequilibrium information
             day_return = 0.0
-            if self.pairs_trades_[pair]:
+            if pair in self.pairs_trades_:
 
                 print(f"{pair}:")
                 pair_return = 0.0
@@ -427,14 +427,16 @@ class BacktestResult:
                         trd["open_scaled_disequilibrium"] is not None
                         and trd["open_scaled_disequilibrium"] is not None
                     ):
-                        disequil_info = f" | Open Dis-eq: {trd['open_scaled_disequilibrium']:.2f},"
-                        f" Close Dis-eq: {trd['open_scaled_disequilibrium']:.2f}"
+                        disequil_info = (
+                            f' | Open Dis-eq: {trd["open_scaled_disequilibrium"]:.2f},'
+                            f' Close Dis-eq: {trd["close_scaled_disequilibrium"]:.2f}'
+                        )
 
                     print(
-                        f" {trd['open_time'].time()}  {trd['symbol']}: "
-                        f" {trd['open_side']} @ ${trd['open_price']:.2f},"
-                        f" {trd["close_side"]} @ ${trd["close_price"]:.2f},"
-                        f" Return: {trd['symbol_return']:.2f}%{disequil_info}"
+                        f' {trd["open_time"].time()}-{trd["close_time"].time()}  {trd["symbol"]}: '
+                        f' {trd["open_side"]} @ ${trd["open_price"]:.2f},'
+                        f' {trd["close_side"]} @ ${trd["close_price"]:.2f},'
+                        f' Return: {trd["symbol_return"]:.2f}%{disequil_info}'
                     )
                     pair_return += trd["symbol_return"]
 
@@ -552,7 +554,7 @@ class BacktestResult:
 
         last_row = pair_result_df.loc[last_row_index]
         last_tstamp = last_row["tstamp"]
-        colname_a, colname_b = pair.colnames()
+        colname_a, colname_b = pair.exec_prices_colnames()
         last_px_a = last_row[colname_a]
         last_px_b = last_row[colname_b]
 
@@ -613,7 +615,7 @@ class BacktestResult:
         return current_value_a, current_value_b, total_current_value
     
     def store_results_in_database(
-        self, db_path: str, datafile: str
+        self, db_path: str, day: str
     ) -> None:
         """
         Store backtest results in the SQLite database.
@@ -623,8 +625,7 @@ class BacktestResult:
 
         try:
             # Extract date from datafile name (assuming format like 20250528.mktdata.ohlcv.db)
-            filename = os.path.basename(datafile)
-            date_str = filename.split(".")[0]  # Extract date part
+            date_str = day
 
             # Convert to proper date format
             try:

lib/pt_trading/rolling_window_fit.py

@@ -146,8 +146,13 @@ class RollingFit(PairsTradingFitMethod):
             print(f"{pair}:  *** Position is NOT CLOSED. ***")
             # outstanding positions
             if config["close_outstanding_positions"]:
+                close_position_row = pd.Series(pair.market_data_.iloc[-2])
+                close_position_row["disequilibrium"] = 0.0
+                close_position_row["scaled_disequilibrium"] = 0.0
+                close_position_row["signed_scaled_disequilibrium"] = 0.0
+                
                 close_position_trades = self._get_close_trades(
-                    pair=pair, row=pred_row, close_threshold=close_threshold
+                    pair=pair, row=close_position_row, close_threshold=close_threshold
                 )
                 if close_position_trades is not None:
                     close_position_trades["status"] = PairState.CLOSE_POSITION.name
@@ -171,9 +176,10 @@ class RollingFit(PairsTradingFitMethod):
     def _get_open_trades(
         self, pair: TradingPair, row: pd.Series, open_threshold: float
     ) -> Optional[pd.DataFrame]:
-        colname_a, colname_b = pair.colnames()
+        colname_a, colname_b = pair.exec_prices_colnames()
 
         open_row = row
+
         open_tstamp = open_row["tstamp"]
         open_disequilibrium = open_row["disequilibrium"]
         open_scaled_disequilibrium = open_row["scaled_disequilibrium"]
@@ -182,7 +188,8 @@ class RollingFit(PairsTradingFitMethod):
         open_px_b = open_row[f"{colname_b}"]
 
         # creating the trades
-        print(f"OPEN_TRADES: {row["tstamp"]} {open_scaled_disequilibrium=}")
+        # use outer single quotes so we can reference DataFrame keys with double quotes inside
+        print(f'OPEN_TRADES: {open_tstamp} open_scaled_disequilibrium={open_scaled_disequilibrium}')
         if open_disequilibrium > 0:
             open_side_a = "SELL"
             open_side_b = "BUY"
@@ -231,10 +238,7 @@ class RollingFit(PairsTradingFitMethod):
             ),
         ]
         # Create DataFrame with explicit dtypes to avoid concatenation warnings
-        df = pd.DataFrame(
-            trd_signal_tuples,
-            columns=self.TRADES_COLUMNS,
-        )
+        df = pd.DataFrame(trd_signal_tuples, columns=self.TRADES_COLUMNS)
         # Ensure consistent dtypes
         return df.astype(
             {
@@ -252,7 +256,7 @@ class RollingFit(PairsTradingFitMethod):
     def _get_close_trades(
         self, pair: TradingPair, row: pd.Series, close_threshold: float
     ) -> Optional[pd.DataFrame]:
-        colname_a, colname_b = pair.colnames()
+        colname_a, colname_b = pair.exec_prices_colnames()
 
         close_row = row
         close_tstamp = close_row["tstamp"]

lib/pt_trading/trading_pair.py

@@ -17,7 +17,7 @@ class PairState(Enum):
 
 class CointegrationData:
     EG_PVALUE_THRESHOLD = 0.05
-    
+
     tstamp_: pd.Timestamp
     pair_: str
     eg_pvalue_: float
@@ -63,7 +63,7 @@ class CointegrationData:
             "johansen_cvt": self.johansen_cvt_,
             "eg_is_cointegrated": self.eg_is_cointegrated_,
             "johansen_is_cointegrated": self.johansen_is_cointegrated_,
-        }   
+        }
 
     def __repr__(self) -> str:
         return f"CointegrationData(tstamp={self.tstamp_}, pair={self.pair_}, eg_pvalue={self.eg_pvalue_}, johansen_lr1={self.johansen_lr1_}, johansen_cvt={self.johansen_cvt_}, eg_is_cointegrated={self.eg_is_cointegrated_}, johansen_is_cointegrated={self.johansen_is_cointegrated_})"
@@ -73,7 +73,7 @@ class TradingPair(ABC):
     market_data_: pd.DataFrame
     symbol_a_: str
     symbol_b_: str
-    price_column_: str
+    stat_model_price_: str
 
     training_mu_: float
     training_std_: float
@@ -86,47 +86,76 @@ class TradingPair(ABC):
     # predicted_df_: Optional[pd.DataFrame]
 
     def __init__(
-        self, config: Dict[str, Any], market_data: pd.DataFrame, symbol_a: str, symbol_b: str, price_column: str
+        self,
+        config: Dict[str, Any],
+        market_data: pd.DataFrame,
+        symbol_a: str,
+        symbol_b: str,
     ):
         self.symbol_a_ = symbol_a
         self.symbol_b_ = symbol_b
-        self.price_column_ = price_column
-        self.set_market_data(market_data)
+        self.stat_model_price_ = config["stat_model_price"]
         self.user_data_ = {}
         self.predicted_df_ = None
         self.config_ = config
 
-    def set_market_data(self, market_data: pd.DataFrame) -> None:
+        self._set_market_data(market_data)
+
+    def _set_market_data(self, market_data: pd.DataFrame) -> None:
         self.market_data_ = pd.DataFrame(
             self._transform_dataframe(market_data)[["tstamp"] + self.colnames()]
         )
 
         self.market_data_ = self.market_data_.dropna().reset_index(drop=True)
-        self.market_data_['tstamp'] = pd.to_datetime(self.market_data_['tstamp'])
-        self.market_data_ = self.market_data_.sort_values('tstamp')
+        self.market_data_["tstamp"] = pd.to_datetime(self.market_data_["tstamp"])
+        self.market_data_ = self.market_data_.sort_values("tstamp")
+        self._set_execution_price_data()
+        pass
+        
+    def _set_execution_price_data(self) -> None:
+        if "execution_price" not in self.config_:
+            self.market_data_[f"exec_price_{self.symbol_a_}"] = self.market_data_[f"{self.stat_model_price_}_{self.symbol_a_}"]
+            self.market_data_[f"exec_price_{self.symbol_b_}"] = self.market_data_[f"{self.stat_model_price_}_{self.symbol_b_}"]
+            return
+        execution_price_column = self.config_["execution_price"]["column"]
+        execution_price_shift = self.config_["execution_price"]["shift"]
+        self.market_data_[f"exec_price_{self.symbol_a_}"] = self.market_data_[f"{self.stat_model_price_}_{self.symbol_a_}"].shift(-execution_price_shift)
+        self.market_data_[f"exec_price_{self.symbol_b_}"] = self.market_data_[f"{self.stat_model_price_}_{self.symbol_b_}"].shift(-execution_price_shift)
+        self.market_data_ = self.market_data_.dropna().reset_index(drop=True)
+
+        
+        
 
     def get_begin_index(self) -> int:
         if "trading_hours" not in self.config_:
             return 0
-        assert "timezone" in self.config_["trading_hours"] 
+        assert "timezone" in self.config_["trading_hours"]
         assert "begin_session" in self.config_["trading_hours"]
-        start_time = pd.to_datetime(self.config_["trading_hours"]["begin_session"]).tz_localize(self.config_["trading_hours"]["timezone"]).time()
-        mask = self.market_data_['tstamp'].dt.time >= start_time
+        start_time = (
+            pd.to_datetime(self.config_["trading_hours"]["begin_session"])
+            .tz_localize(self.config_["trading_hours"]["timezone"])
+            .time()
+        )
+        mask = self.market_data_["tstamp"].dt.time >= start_time
         return int(self.market_data_.index[mask].min())
 
     def get_end_index(self) -> int:
         if "trading_hours" not in self.config_:
             return 0
-        assert "timezone" in self.config_["trading_hours"] 
+        assert "timezone" in self.config_["trading_hours"]
         assert "end_session" in self.config_["trading_hours"]
-        end_time = pd.to_datetime(self.config_["trading_hours"]["end_session"]).tz_localize(self.config_["trading_hours"]["timezone"]).time()
-        mask = self.market_data_['tstamp'].dt.time <= end_time
+        end_time = (
+            pd.to_datetime(self.config_["trading_hours"]["end_session"])
+            .tz_localize(self.config_["trading_hours"]["timezone"])
+            .time()
+        )
+        mask = self.market_data_["tstamp"].dt.time <= end_time
         return int(self.market_data_.index[mask].max())
 
     def _transform_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
         # Select only the columns we need
         df_selected: pd.DataFrame = pd.DataFrame(
-            df[["tstamp", "symbol", self.price_column_]]
+            df[["tstamp", "symbol", self.stat_model_price_]]
         )
 
         # Start with unique timestamps
@@ -144,13 +173,13 @@ class TradingPair(ABC):
             )
 
             # Create column name like "close-COIN"
-            new_price_column = f"{self.price_column_}_{symbol}"
+            new_price_column = f"{self.stat_model_price_}_{symbol}"
 
             # Create temporary dataframe with timestamp and price
             temp_df = pd.DataFrame(
                 {
                     "tstamp": df_symbol["tstamp"],
-                    new_price_column: df_symbol[self.price_column_],
+                    new_price_column: df_symbol[self.stat_model_price_],
                 }
             )
 
@@ -171,7 +200,7 @@ class TradingPair(ABC):
 
         testing_start_index = training_start_index + training_minutes
         self.training_df_ = self.market_data_.iloc[
-            training_start_index:testing_start_index, : training_minutes
+            training_start_index:testing_start_index, :training_minutes
         ].copy()
         assert self.training_df_ is not None
         self.training_df_ = self.training_df_.dropna().reset_index(drop=True)
@@ -188,8 +217,14 @@ class TradingPair(ABC):
 
     def colnames(self) -> List[str]:
         return [
-            f"{self.price_column_}_{self.symbol_a_}",
-            f"{self.price_column_}_{self.symbol_b_}",
+            f"{self.stat_model_price_}_{self.symbol_a_}",
+            f"{self.stat_model_price_}_{self.symbol_b_}",
+        ]
+
+    def exec_prices_colnames(self) -> List[str]:
+        return [
+            f"exec_price_{self.symbol_a_}",
+            f"exec_price_{self.symbol_b_}",
         ]
 
     def add_trades(self, trades: pd.DataFrame) -> None:
@@ -199,7 +234,7 @@ class TradingPair(ABC):
         else:
             # Ensure both DataFrames have the same columns and dtypes before concatenation
             existing_trades = self.user_data_["trades"]
-            
+
             # If existing trades is empty, just assign the new trades
             if len(existing_trades) == 0:
                 self.user_data_["trades"] = trades.copy()
@@ -213,22 +248,26 @@ class TradingPair(ABC):
                                 trades[col] = pd.Timestamp.now()
                             elif col in ["action", "symbol"]:
                                 trades[col] = ""
-                            elif col in ["price", "disequilibrium", "scaled_disequilibrium"]:
+                            elif col in [
+                                "price",
+                                "disequilibrium",
+                                "scaled_disequilibrium",
+                            ]:
                                 trades[col] = 0.0
                             elif col == "pair":
                                 trades[col] = None
                             else:
                                 trades[col] = None
-                
+
                 # Concatenate with explicit dtypes to avoid warnings
                 self.user_data_["trades"] = pd.concat(
-                    [existing_trades, trades], 
-                    ignore_index=True,
-                    copy=False
+                    [existing_trades, trades], ignore_index=True, copy=False
                 )
 
     def get_trades(self) -> pd.DataFrame:
-        return self.user_data_["trades"] if "trades" in self.user_data_ else pd.DataFrame()
+        return (
+            self.user_data_["trades"] if "trades" in self.user_data_ else pd.DataFrame()
+        )
 
     def cointegration_check(self) -> Optional[pd.DataFrame]:
         print(f"***{self}*** STARTING....")
@@ -237,17 +276,19 @@ class TradingPair(ABC):
         curr_training_start_idx = 0
 
         COINTEGRATION_DATA_COLUMNS = {
-            "tstamp" : "datetime64[ns]",
-            "pair" : "string",
-            "eg_pvalue" : "float64",
-            "johansen_lr1" : "float64",
-            "johansen_cvt" : "float64",
-            "eg_is_cointegrated" : "bool",
-            "johansen_is_cointegrated" : "bool",
+            "tstamp": "datetime64[ns]",
+            "pair": "string",
+            "eg_pvalue": "float64",
+            "johansen_lr1": "float64",
+            "johansen_cvt": "float64",
+            "eg_is_cointegrated": "bool",
+            "johansen_is_cointegrated": "bool",
         }
         # Initialize trades DataFrame with proper dtypes to avoid concatenation warnings
-        result: pd.DataFrame = pd.DataFrame(columns=[col for col in COINTEGRATION_DATA_COLUMNS.keys()]) #.astype(COINTEGRATION_DATA_COLUMNS)
-        
+        result: pd.DataFrame = pd.DataFrame(
+            columns=[col for col in COINTEGRATION_DATA_COLUMNS.keys()]
+        )  # .astype(COINTEGRATION_DATA_COLUMNS)
+
         training_minutes = config["training_minutes"]
         while True:
             print(curr_training_start_idx, end="\r")
@@ -271,13 +312,16 @@ class TradingPair(ABC):
 
     def to_stop_close_conditions(self, predicted_row: pd.Series) -> bool:
         config = self.config_
-        if ("stop_close_conditions" not in config or config["stop_close_conditions"] is None) :
+        if (
+            "stop_close_conditions" not in config
+            or config["stop_close_conditions"] is None
+        ):
             return False
         if "profit" in config["stop_close_conditions"]:
             current_return = self._current_return(predicted_row)
             #
             # print(f"time={predicted_row['tstamp']} current_return={current_return}")
-            # 
+            #
             if current_return >= config["stop_close_conditions"]["profit"]:
                 print(f"STOP PROFIT: {current_return}")
                 self.user_data_["stop_close_state"] = PairState.CLOSE_STOP_PROFIT
@@ -288,9 +332,10 @@ class TradingPair(ABC):
                 self.user_data_["stop_close_state"] = PairState.CLOSE_STOP_LOSS
                 return True
         return False
-    
+
     def on_open_trades(self, trades: pd.DataFrame) -> None:
-        if "close_trades" in self.user_data_: del self.user_data_["close_trades"]
+        if "close_trades" in self.user_data_:
+            del self.user_data_["close_trades"]
         self.user_data_["open_trades"] = trades
 
     def on_close_trades(self, trades: pd.DataFrame) -> None:
@@ -302,20 +347,25 @@ class TradingPair(ABC):
             open_trades = self.user_data_["open_trades"]
             if len(open_trades) == 0:
                 return 0.0
+
             def _single_instrument_return(symbol: str) -> float:
                 instrument_open_trades = open_trades[open_trades["symbol"] == symbol]
                 instrument_open_price = instrument_open_trades["price"].iloc[0]
 
-                sign = -1 if instrument_open_trades["side"].iloc[0]  == "SELL" else 1
-                instrument_price = predicted_row[f"{self.price_column_}_{symbol}"]
-                instrument_return = sign * (instrument_price - instrument_open_price) / instrument_open_price
+                sign = -1 if instrument_open_trades["side"].iloc[0] == "SELL" else 1
+                instrument_price = predicted_row[f"{self.stat_model_price_}_{symbol}"]
+                instrument_return = (
+                    sign
+                    * (instrument_price - instrument_open_price)
+                    / instrument_open_price
+                )
                 return float(instrument_return) * 100.0
-            
+
             instrument_a_return = _single_instrument_return(self.symbol_a_)
             instrument_b_return = _single_instrument_return(self.symbol_b_)
-            return (instrument_a_return + instrument_b_return)
+            return instrument_a_return + instrument_b_return
         return 0.0
-        
+
     def __repr__(self) -> str:
         return self.name()
 
@@ -328,4 +378,3 @@ class TradingPair(ABC):
 
     # @abstractmethod
     # def predicted_df(self) -> Optional[pd.DataFrame]: ...
-

lib/pt_trading/vecm_rolling_fit.py

@@ -1,34 +1,78 @@
+# original script moved to vecm_rolling_fit_01.py
+
+# 09.09.25  Added GARCH model - predicting volatility
+
+# Rule of thumb:
+# alpha + beta ≈ 1 → strong volatility clustering, persistence.
+# If much lower → volatility mean reverts quickly.
+# If > 1 → model is unstable / non-stationary (bad).
+
+# the VECM disequilibrium (mean reversion signal) and 
+# the GARCH volatility forecast (risk measure).
+# combine them → e.g., only enter trades when:
+
+# high_volatility = 1 → persistence > 0.95 or volatility > 2 (rule of thumb: unstable / risky regime).
+# high_volatility = 0 → stable regime.
+
+
+# VECM disequilibrium z-score > threshold and
+# GARCH-forecasted volatility is not too high (avoid noise-driven signals).
+# This creates a volatility-adjusted pairs trading strategy, more robust than plain VECM
+
+# now pair_predict_result_ DataFrame includes:
+# disequilibrium, scaled_disequilibrium, z-scores, garch_alpha, garch_beta, garch_persistence (α+β rule-of-thumb)
+# garch_vol_forecast (1-step volatility forecast)
+
+# Would you like me to also add a warning flag column 
+# (e.g., "high_volatility" = 1 if persistence > 0.95 or vol_forecast > threshold) 
+# so you can easily detect unstable regimes?
+
+# VECM/GARCH 
+# vecm_rolling_fit.py:
 from typing import Any, Dict, Optional, cast
 
+import numpy as np
 import pandas as pd
+from typing import Any, Dict, Optional
 from pt_trading.results import BacktestResult
 from pt_trading.rolling_window_fit import RollingFit
 from pt_trading.trading_pair import TradingPair
 from statsmodels.tsa.vector_ar.vecm import VECM, VECMResults
+from arch import arch_model
 
 NanoPerMin = 1e9
+
 class VECMTradingPair(TradingPair):
     vecm_fit_: Optional[VECMResults]
     pair_predict_result_: Optional[pd.DataFrame]
-    
-    def __init__(self, config: Dict[str, Any], market_data: pd.DataFrame, symbol_a: str, symbol_b: str, price_column: str):
-        super().__init__(config, market_data, symbol_a, symbol_b, price_column)
+
+    def __init__(
+        self,
+        config: Dict[str, Any],
+        market_data: pd.DataFrame,
+        symbol_a: str,
+        symbol_b: str,
+    ):
+        super().__init__(config, market_data, symbol_a, symbol_b)
         self.vecm_fit_ = None
         self.pair_predict_result_ = None
-        
+        self.garch_fit_ = None
+        self.sigma_spread_forecast_ = None
+        self.garch_alpha_ = None
+        self.garch_beta_ = None
+        self.garch_persistence_ = None
+        self.high_volatility_flag_ = None
+
     def _train_pair(self) -> None:
         self._fit_VECM()
         assert self.vecm_fit_ is not None
+
         diseq_series = self.training_df_[self.colnames()] @ self.vecm_fit_.beta
-        # print(diseq_series.shape)
         self.training_mu_ = float(diseq_series[0].mean())
         self.training_std_ = float(diseq_series[0].std())
 
-        self.training_df_["dis-equilibrium"] = (
-            self.training_df_[self.colnames()] @ self.vecm_fit_.beta
-        )
-        # Normalize the dis-equilibrium
-        self.training_df_["scaled_dis-equilibrium"] = (
+        self.training_df_["disequilibrium"] = diseq_series
+        self.training_df_["scaled_disequilibrium"] = (
             diseq_series - self.training_mu_
         ) / self.training_std_
 
@@ -37,73 +81,113 @@ class VECMTradingPair(TradingPair):
         vecm_df = self.training_df_[self.colnames()].reset_index(drop=True)
         vecm_model = VECM(vecm_df, coint_rank=1)
         vecm_fit = vecm_model.fit()
-
-        assert vecm_fit is not None
-
-        # URGENT check beta and alpha
-
-        # Check if the model converged properly
-        if not hasattr(vecm_fit, "beta") or vecm_fit.beta is None:
-            print(f"{self}: VECM model failed to converge properly")
-
         self.vecm_fit_ = vecm_fit
-        pass
+
+        # Error Correction Term (spread)
+        ect_series = (vecm_df @ vecm_fit.beta).iloc[:, 0]
+
+        # Difference the spread for stationarity
+        dz = ect_series.diff().dropna()
+
+        if len(dz) < 30:
+            print("Not enough data for GARCH fitting.")
+            return
+
+        # Rescale if variance too small
+        if dz.std() < 0.1:
+            dz = dz * 1000
+        # print("Scale check:", dz.std())
+
+        try:
+            garch = arch_model(dz, vol="GARCH", p=1, q=1, mean="Zero", dist="normal")
+            garch_fit = garch.fit(disp="off")
+            self.garch_fit_ = garch_fit
+
+            # Extract parameters
+            params = garch_fit.params
+            self.garch_alpha_ = params.get("alpha[1]", np.nan)
+            self.garch_beta_ = params.get("beta[1]", np.nan)
+            self.garch_persistence_ = self.garch_alpha_ + self.garch_beta_
+
+            # print (f"GARCH α: {self.garch_alpha_:.4f}, β: {self.garch_beta_:.4f}, "
+            #       f"α+β (persistence): {self.garch_persistence_:.4f}")
+
+            # One-step-ahead volatility forecast
+            forecast = garch_fit.forecast(horizon=1)
+            sigma_next = np.sqrt(forecast.variance.iloc[-1, 0])
+            self.sigma_spread_forecast_ = float(sigma_next)
+            # print("GARCH sigma forecast:", self.sigma_spread_forecast_)
+
+            # Rule of thumb: persistence close to 1 or large volatility forecast
+            self.high_volatility_flag_ = int(
+                (self.garch_persistence_ is not None and self.garch_persistence_ > 0.95)
+                or (self.sigma_spread_forecast_ is not None and self.sigma_spread_forecast_ > 2)
+            )
+
+        except Exception as e:
+            print(f"GARCH fit failed: {e}")
+            self.garch_fit_ = None
+            self.sigma_spread_forecast_ = None
+            self.high_volatility_flag_ = None
 
     def predict(self) -> pd.DataFrame:
         self._train_pair()
-        
         assert self.testing_df_ is not None
         assert self.vecm_fit_ is not None
+
+        # VECM predictions
         predicted_prices = self.vecm_fit_.predict(steps=len(self.testing_df_))
-
-        # Convert prediction to a DataFrame for readability
-        predicted_df = pd.DataFrame(
-            predicted_prices, columns=pd.Index(self.colnames()), dtype=float
-        )
-
         predicted_df = pd.merge(
             self.testing_df_.reset_index(drop=True),
-            pd.DataFrame(
-                predicted_prices, columns=pd.Index(self.colnames()), dtype=float
-            ),
+            pd.DataFrame(predicted_prices, columns=pd.Index(self.colnames()), dtype=float),
             left_index=True,
             right_index=True,
             suffixes=("", "_pred"),
         ).dropna()
 
+        # Disequilibrium and z-scores
         predicted_df["disequilibrium"] = (
             predicted_df[self.colnames()] @ self.vecm_fit_.beta
         )
-
         predicted_df["signed_scaled_disequilibrium"] = (
             predicted_df["disequilibrium"] - self.training_mu_
         ) / self.training_std_
-
-        predicted_df["scaled_disequilibrium"] = (
-            abs(predicted_df["signed_scaled_disequilibrium"])
+        predicted_df["scaled_disequilibrium"] = abs(
+            predicted_df["signed_scaled_disequilibrium"]
         )
-        
-        predicted_df = predicted_df.reset_index(drop=True)
+
+        # Add GARCH parameters + volatility forecast
+        predicted_df["garch_alpha"] = self.garch_alpha_
+        predicted_df["garch_beta"] = self.garch_beta_
+        predicted_df["garch_persistence"] = self.garch_persistence_
+        predicted_df["garch_vol_forecast"] = self.sigma_spread_forecast_
+        predicted_df["high_volatility"] = self.high_volatility_flag_
+
+        # Save results
         if self.pair_predict_result_ is None:
             self.pair_predict_result_ = predicted_df
         else:
-            self.pair_predict_result_ = pd.concat([self.pair_predict_result_, predicted_df], ignore_index=True)
-        # Reset index to ensure proper indexing 
-        self.pair_predict_result_ = self.pair_predict_result_.reset_index(drop=True)
+            self.pair_predict_result_ = pd.concat(
+                [self.pair_predict_result_, predicted_df], ignore_index=True
+            )
+
         return self.pair_predict_result_
 
-    
+
 class VECMRollingFit(RollingFit):
     def __init__(self) -> None:
         super().__init__()
 
     def create_trading_pair(
-        self, config: Dict, market_data: pd.DataFrame, symbol_a: str, symbol_b: str, price_column: str
+        self,
+        config: Dict,
+        market_data: pd.DataFrame,
+        symbol_a: str,
+        symbol_b: str,
     ) -> TradingPair:
         return VECMTradingPair(
             config=config,
             market_data=market_data,
-            symbol_a=symbol_a,
-            symbol_b=symbol_b,
-            price_column=price_column
-        )
+            symbol_a = symbol_a,
+            symbol_b = symbol_b,
+        )

lib/pt_trading/z-score_rolling_fit.py

@@ -1,74 +1,124 @@
-from typing import Any, Dict, Optional, cast
+from typing import Any, Dict, Optional
 
 import pandas as pd
-from pt_trading.results import BacktestResult
-from pt_trading.rolling_window_fit import RollingFit
-from pt_trading.trading_pair import TradingPair
 import statsmodels.api as sm
 
+from pt_trading.rolling_window_fit import RollingFit
+from pt_trading.trading_pair import TradingPair
+
 NanoPerMin = 1e9
+
+
 class ZScoreTradingPair(TradingPair):
+    """TradingPair implementation that fits a hedge ratio with OLS and
+    computes a standardized spread (z-score).
+
+    The class stores training spread mean/std and hedge ratio so the model
+    can be applied to testing data consistently.
+    """
+
     zscore_model_: Optional[sm.regression.linear_model.RegressionResultsWrapper]
     pair_predict_result_: Optional[pd.DataFrame]
-    zscore_df_: Optional[pd.DataFrame]
-    
-    def __init__(self, config: Dict[str, Any], market_data: pd.DataFrame, symbol_a: str, symbol_b: str, price_column: str):
-        super().__init__(config, market_data, symbol_a, symbol_b, price_column)
+    zscore_df_: Optional[pd.Series]
+    hedge_ratio_: Optional[float]
+    spread_mean_: Optional[float]
+    spread_std_: Optional[float]
+
+    def __init__(
+        self,
+        config: Dict[str, Any],
+        market_data: pd.DataFrame,
+        symbol_a: str,
+        symbol_b: str,
+    ):
+        super().__init__(config, market_data, symbol_a, symbol_b)
         self.zscore_model_ = None
         self.pair_predict_result_ = None
         self.zscore_df_ = None
-        
-    def _fit_zscore(self) -> None:
-        assert self.training_df_ is not None
-        symbol_a_px_series = self.training_df_[self.colnames()].iloc[:, 0]
-        symbol_b_px_series = self.training_df_[self.colnames()].iloc[:, 1]
-        
-        symbol_a_px_series,symbol_b_px_series = symbol_a_px_series.align(symbol_b_px_series, axis=0)
-        
-        X = sm.add_constant(symbol_b_px_series)
-        self.zscore_model_ = sm.OLS(symbol_a_px_series, X).fit()
-        assert self.zscore_model_ is not None
-        hedge_ratio = self.zscore_model_.params.iloc[1]
+        self.hedge_ratio_ = None
+        self.spread_mean_ = None
+        self.spread_std_ = None
 
-        # Calculate spread and Z-score
-        spread = symbol_a_px_series - hedge_ratio * symbol_b_px_series
-        self.zscore_df_  = (spread - spread.mean()) / spread.std()        
+    def _fit_zscore(self) -> None:
+        """Fit OLS on the training window and compute training z-score."""
+        assert self.training_df_ is not None
+
+        # Extract price series for the two symbols from the training frame.
+        px_df = self.training_df_[self.colnames()]
+        symbol_a_px = px_df.iloc[:, 0]
+        symbol_b_px = px_df.iloc[:, 1]
+
+        # Align indexes and fit OLS: symbol_a ~ const + symbol_b
+        symbol_a_px, symbol_b_px = symbol_a_px.align(symbol_b_px, join="inner")
+        X = sm.add_constant(symbol_b_px)
+        self.zscore_model_ = sm.OLS(symbol_a_px, X).fit()
+
+        # Hedge ratio is the slope on symbol_b
+        params = self.zscore_model_.params
+        self.hedge_ratio_ = float(params.iloc[1]) if len(params) > 1 else 0.0
+
+        # Training spread and its standardized z-score
+        spread = symbol_a_px - self.hedge_ratio_ * symbol_b_px
+        self.spread_mean_ = float(spread.mean())
+        self.spread_std_ = float(spread.std(ddof=0)) if spread.std(ddof=0) != 0 else 1.0
+        self.zscore_df_ = (spread - self.spread_mean_) / self.spread_std_
 
     def predict(self) -> pd.DataFrame:
+        """Apply fitted hedge ratio to the testing frame and return a
+        dataframe with canonical columns:
+          - disequilibrium: signed z-score
+          - scaled_disequilibrium: absolute z-score
+          - signed_scaled_disequilibrium: same as disequilibrium (keeps sign)
+        """
+        # Fit on training window
         self._fit_zscore()
         assert self.zscore_df_ is not None
-        self.training_df_["dis-equilibrium"] = self.zscore_df_
-        self.training_df_["scaled_dis-equilibrium"] = abs(self.zscore_df_)
-        
-        assert self.testing_df_ is not None
-        assert self.zscore_df_ is not None
-        predicted_df = self.testing_df_
+        assert self.hedge_ratio_ is not None
+        assert self.spread_mean_ is not None and self.spread_std_ is not None
 
-        predicted_df["disequilibrium"] = self.zscore_df_
-        predicted_df["signed_scaled_disequilibrium"] = self.zscore_df_
-        predicted_df["scaled_disequilibrium"] = abs(self.zscore_df_)
-        
-        predicted_df = predicted_df.reset_index(drop=True)
+        # Keep training columns for inspection
+        self.training_df_["disequilibrium"] = self.zscore_df_
+        self.training_df_["scaled_disequilibrium"] = self.zscore_df_.abs()
+
+        # Apply model to testing frame
+        assert self.testing_df_ is not None
+        test_df = self.testing_df_.copy()
+        px_test = test_df[self.colnames()]
+        a_test = px_test.iloc[:, 0]
+        b_test = px_test.iloc[:, 1]
+        a_test, b_test = a_test.align(b_test, join="inner")
+
+        # Compute test spread and standardize using training mean/std
+        test_spread = a_test - self.hedge_ratio_ * b_test
+        test_zscore = (test_spread - self.spread_mean_) / self.spread_std_
+
+        # Attach canonical columns
+        # Align back to test_df index if needed
+        test_zscore = test_zscore.reindex(test_df.index)
+        test_df["disequilibrium"] = test_zscore
+        test_df["signed_scaled_disequilibrium"] = test_zscore
+        test_df["scaled_disequilibrium"] = test_zscore.abs()
+
+        # Reset index and accumulate results across windows
+        test_df = test_df.reset_index(drop=True)
         if self.pair_predict_result_ is None:
-            self.pair_predict_result_ = predicted_df
+            self.pair_predict_result_ = test_df
         else:
-            self.pair_predict_result_ = pd.concat([self.pair_predict_result_, predicted_df], ignore_index=True)
-        # Reset index to ensure proper indexing 
+            self.pair_predict_result_ = pd.concat(
+                [self.pair_predict_result_, test_df], ignore_index=True
+            )
+
         self.pair_predict_result_ = self.pair_predict_result_.reset_index(drop=True)
         return self.pair_predict_result_.dropna()
 
-    
+
 class ZScoreRollingFit(RollingFit):
     def __init__(self) -> None:
         super().__init__()
 
     def create_trading_pair(
-        self, config: Dict, market_data: pd.DataFrame, symbol_a: str, symbol_b: str, price_column: str
+        self, config: Dict, market_data: pd.DataFrame, symbol_a: str, symbol_b: str
     ) -> TradingPair:
         return ZScoreTradingPair(
-            config=config,
-            market_data=market_data,
-            symbol_a=symbol_a,
-            symbol_b=symbol_b,
-            price_column=price_column
+            config=config, market_data=market_data, symbol_a=symbol_a, symbol_b=symbol_b
         )

lib/tools/data_loader.py

@@ -5,7 +5,13 @@ from typing import Dict, List, cast
 import pandas as pd
 
 
-def load_sqlite_to_dataframe(db_path, query):
+def load_sqlite_to_dataframe(db_path:str, query:str) -> pd.DataFrame:
+    df: pd.DataFrame = pd.DataFrame()
+    import os
+    if not os.path.exists(db_path):
+        print(f"WARNING: database file {db_path} does not exist")
+        return df
+
     try:
         conn = sqlite3.connect(db_path)
 
@@ -22,13 +28,14 @@ def load_sqlite_to_dataframe(db_path, query):
             conn.close()
 
 
-def convert_time_to_UTC(value: str, timezone: str) -> str:
+def convert_time_to_UTC(value: str, timezone: str, extra_minutes: int = 0) -> str:
 
     from zoneinfo import ZoneInfo
-    from datetime import datetime
+    from datetime import datetime, timedelta
 
     # Parse it to naive datetime object
     local_dt = datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
+    local_dt = local_dt + timedelta(minutes=extra_minutes)
 
     zinfo = ZoneInfo(timezone)
     result: datetime = local_dt.replace(tzinfo=zinfo).astimezone(ZoneInfo("UTC"))
@@ -41,6 +48,7 @@ def load_market_data(
     instruments: List[Dict[str, str]],
     db_table_name: str,
     trading_hours: Dict = {},
+    extra_minutes: int = 0,
 ) -> pd.DataFrame:
 
     insts = [
@@ -79,7 +87,7 @@ def load_market_data(
             f"{date_str} {trading_hours['begin_session']}", trading_hours["timezone"]
         )
         end_time = convert_time_to_UTC(
-            f"{date_str} {trading_hours['end_session']}", trading_hours["timezone"]
+            f"{date_str} {trading_hours['end_session']}", trading_hours["timezone"], extra_minutes=extra_minutes # to get execution price
         )
 
         # Perform boolean selection

requirements.txt

@@ -61,7 +61,7 @@ protobuf>=3.12.4
 psutil>=5.9.0
 ptyprocess>=0.7.0
 pycurl>=7.44.1
-pyelftools>=0.27
+# pyelftools>=0.27
 Pygments>=2.11.2
 pyparsing>=2.4.7
 pyrsistent>=0.18.1
@@ -69,7 +69,7 @@ python-debian>=0.1.43 #+ubuntu1.1
 python-dotenv>=0.19.2
 python-magic>=0.4.24
 python-xlib>=0.29
-pyxdg>=0.27
+# pyxdg>=0.27
 PyYAML>=6.0
 reportlab>=3.6.8
 requests>=2.25.1
@@ -82,113 +82,113 @@ six>=1.16.0
 soupsieve>=2.3.1
 ssh-import-id>=5.11
 statsmodels>=0.14.4
-texttable>=1.6.4
+# texttable>=1.6.4
 tldextract>=3.1.2
 tomli>=1.2.2
 ######## typed-ast>=1.4.3
-types-aiofiles>=0.1
-types-annoy>=1.17
-types-appdirs>=1.4
-types-atomicwrites>=1.4
-types-aws-xray-sdk>=2.8
-types-babel>=2.9
-types-backports-abc>=0.5
-types-backports.ssl-match-hostname>=3.7
-types-beautifulsoup4>=4.10
-types-bleach>=4.1
-types-boto>=2.49
-types-braintree>=4.11
-types-cachetools>=4.2
-types-caldav>=0.8
-types-certifi>=2020.4
-types-characteristic>=14.3
-types-chardet>=4.0
-types-click>=7.1
-types-click-spinner>=0.1
-types-colorama>=0.4
-types-commonmark>=0.9
-types-contextvars>=0.1
-types-croniter>=1.0
-types-cryptography>=3.3
-types-dataclasses>=0.1
-types-dateparser>=1.0
-types-DateTimeRange>=0.1
-types-decorator>=0.1
-types-Deprecated>=1.2
-types-docopt>=0.6
-types-docutils>=0.17
-types-editdistance>=0.5
-types-emoji>=1.2
-types-entrypoints>=0.3
-types-enum34>=1.1
-types-filelock>=3.2
-types-first>=2.0
-types-Flask>=1.1
-types-freezegun>=1.1
-types-frozendict>=0.1
-types-futures>=3.3
-types-html5lib>=1.1
-types-httplib2>=0.19
-types-humanfriendly>=9.2
-types-ipaddress>=1.0
-types-itsdangerous>=1.1
-types-JACK-Client>=0.1
-types-Jinja2>=2.11
-types-jmespath>=0.10
-types-jsonschema>=3.2
-types-Markdown>=3.3
-types-MarkupSafe>=1.1
-types-mock>=4.0
-types-mypy-extensions>=0.4
-types-mysqlclient>=2.0
-types-oauthlib>=3.1
-types-orjson>=3.6
-types-paramiko>=2.7
-types-Pillow>=8.3
-types-polib>=1.1
-types-prettytable>=2.1
-types-protobuf>=3.17
-types-psutil>=5.8
-types-psycopg2>=2.9
-types-pyaudio>=0.2
-types-pycurl>=0.1
-types-pyfarmhash>=0.2
-types-Pygments>=2.9
-types-PyMySQL>=1.0
-types-pyOpenSSL>=20.0
-types-pyRFC3339>=0.1
-types-pysftp>=0.2
-types-pytest-lazy-fixture>=0.6
-types-python-dateutil>=2.8
-types-python-gflags>=3.1
-types-python-nmap>=0.6
-types-python-slugify>=5.0
-types-pytz>=2021.1
-types-pyvmomi>=7.0
-types-PyYAML>=5.4
-types-redis>=3.5
-types-requests>=2.25
-types-retry>=0.9
-types-selenium>=3.141
-types-Send2Trash>=1.8
-types-setuptools>=57.4
-types-simplejson>=3.17
-types-singledispatch>=3.7
-types-six>=1.16
-types-slumber>=0.7
-types-stripe>=2.59
-types-tabulate>=0.8
-types-termcolor>=1.1
-types-toml>=0.10
-types-toposort>=1.6
-types-ttkthemes>=3.2
-types-typed-ast>=1.4
-types-tzlocal>=0.1
-types-ujson>=0.1
-types-vobject>=0.9
-types-waitress>=0.1
-types-Werkzeug>=1.0
-types-xxhash>=2.0
+# types-aiofiles>=0.1
+# types-annoy>=1.17
+# types-appdirs>=1.4
+# types-atomicwrites>=1.4
+# types-aws-xray-sdk>=2.8
+# types-babel>=2.9
+# types-backports-abc>=0.5
+# types-backports.ssl-match-hostname>=3.7
+# types-beautifulsoup4>=4.10
+# types-bleach>=4.1
+# types-boto>=2.49
+# types-braintree>=4.11
+# types-cachetools>=4.2
+# types-caldav>=0.8
+# types-certifi>=2020.4
+# types-characteristic>=14.3
+# types-chardet>=4.0
+# types-click>=7.1
+# types-click-spinner>=0.1
+# types-colorama>=0.4
+# types-commonmark>=0.9
+# types-contextvars>=0.1
+# types-croniter>=1.0
+# types-cryptography>=3.3
+# types-dataclasses>=0.1
+# types-dateparser>=1.0
+# types-DateTimeRange>=0.1
+# types-decorator>=0.1
+# types-Deprecated>=1.2
+# types-docopt>=0.6
+# types-docutils>=0.17
+# types-editdistance>=0.5
+# types-emoji>=1.2
+# types-entrypoints>=0.3
+# types-enum34>=1.1
+# types-filelock>=3.2
+# types-first>=2.0
+# types-Flask>=1.1
+# types-freezegun>=1.1
+# types-frozendict>=0.1
+# types-futures>=3.3
+# types-html5lib>=1.1
+# types-httplib2>=0.19
+# types-humanfriendly>=9.2
+# types-ipaddress>=1.0
+# types-itsdangerous>=1.1
+# types-JACK-Client>=0.1
+# types-Jinja2>=2.11
+# types-jmespath>=0.10
+# types-jsonschema>=3.2
+# types-Markdown>=3.3
+# types-MarkupSafe>=1.1
+# types-mock>=4.0
+# types-mypy-extensions>=0.4
+# types-mysqlclient>=2.0
+# types-oauthlib>=3.1
+# types-orjson>=3.6
+# types-paramiko>=2.7
+# types-Pillow>=8.3
+# types-polib>=1.1
+# types-prettytable>=2.1
+# types-protobuf>=3.17
+# types-psutil>=5.8
+# types-psycopg2>=2.9
+# types-pyaudio>=0.2
+# types-pycurl>=0.1
+# types-pyfarmhash>=0.2
+# types-Pygments>=2.9
+# types-PyMySQL>=1.0
+# types-pyOpenSSL>=20.0
+# types-pyRFC3339>=0.1
+# types-pysftp>=0.2
+# types-pytest-lazy-fixture>=0.6
+# types-python-dateutil>=2.8
+# types-python-gflags>=3.1
+# types-python-nmap>=0.6
+# types-python-slugify>=5.0
+# types-pytz>=2021.1
+# types-pyvmomi>=7.0
+# types-PyYAML>=5.4
+# types-redis>=3.5
+# types-requests>=2.25
+# types-retry>=0.9
+# types-selenium>=3.141
+# types-Send2Trash>=1.8
+# types-setuptools>=57.4
+# types-simplejson>=3.17
+# types-singledispatch>=3.7
+# types-six>=1.16
+# types-slumber>=0.7
+# types-stripe>=2.59
+# types-tabulate>=0.8
+# types-termcolor>=1.1
+# types-toml>=0.10
+# types-toposort>=1.6
+# types-ttkthemes>=3.2
+# types-typed-ast>=1.4
+# types-tzlocal>=0.1
+# types-ujson>=0.1
+# types-vobject>=0.9
+# types-waitress>=0.1
+#types-Werkzeug>=1.0
+#types-xxhash>=2.0
 typing-extensions>=3.10.0.2
 Unidecode>=1.3.3
 urllib3>=1.26.5

research/cointegration_test.py

@@ -9,18 +9,19 @@ import pandas as pd
 
 from tools.config import expand_filename, load_config
 from tools.data_loader import get_available_instruments_from_db
+
 from pt_trading.results import (
     BacktestResult,
     create_result_database,
     store_config_in_database,
     store_results_in_database,
 )
+
 from pt_trading.fit_method import PairsTradingFitMethod
 from pt_trading.trading_pair import TradingPair
 
 from research.research_tools import create_pairs, resolve_datafiles
 
-
 def main() -> None:
     parser = argparse.ArgumentParser(description="Run pairs trading backtest.")
     parser.add_argument(
@@ -36,7 +37,7 @@ def main() -> None:
         "--instruments",
         type=str,
         required=False,
-        help="Comma-separated list of instrument symbols (e.g., COIN,GBTC). If not provided, auto-detects from database.",
+        help = "Comma-separated list of instrument symbols (e.g., COIN,GBTC). If not provided, auto-detects from database.",
     )
     args = parser.parse_args()
 
@@ -85,7 +86,7 @@ def main() -> None:
     #     )
 
     # Process each data file
-    price_column = config["price_column"]
+    stat_model_price = config["stat_model_price"]
 
     print(f"\n====== Processing {os.path.basename(datafile)} ======")
 
@@ -105,7 +106,7 @@ def main() -> None:
     # Process data for this file
     try:
         cointegration_data: pd.DataFrame = pd.DataFrame()
-        for pair in create_pairs(datafile, price_column, config, instruments):
+        for pair in create_pairs(datafile, stat_model_price, config, instruments):
             cointegration_data = pd.concat([cointegration_data, pair.cointegration_check()])
 
         pd.set_option('display.width', 400)

research/notebooks/pt_pair_backtest.ipynb

@@ -0,0 +1,16 @@
+{
+ "cells": [],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "name": "python",
+   "version": "3.12.5"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

research/pt_backtest.py

@@ -3,7 +3,7 @@ import glob
 import importlib
 import os
 from datetime import date, datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 import pandas as pd
 
@@ -17,11 +17,13 @@ from pt_trading.results import (
 from pt_trading.fit_method import PairsTradingFitMethod
 from pt_trading.trading_pair import TradingPair
 
+DayT = str
+DataFileNameT = str
 
 def resolve_datafiles(
     config: Dict, date_pattern: str, instruments: List[Dict[str, str]]
-) -> List[str]:
-    resolved_files = []
+) -> List[Tuple[DayT, DataFileNameT]]:
+    resolved_files: List[Tuple[DayT, DataFileNameT]] = []
     for inst in instruments:
         pattern = date_pattern
         inst_type = inst["instrument_type"]
@@ -31,12 +33,17 @@ def resolve_datafiles(
             if not os.path.isabs(pattern):
                 pattern = os.path.join(data_dir, f"{pattern}.mktdata.ohlcv.db")
             matched_files = glob.glob(pattern)
-            resolved_files.extend(matched_files)
+            for matched_file in matched_files:
+                import re
+                match = re.search(r"(\d{8})\.mktdata\.ohlcv\.db$", matched_file)
+                assert match is not None
+                day = match.group(1)
+                resolved_files.append((day, matched_file))
         else:
             # Handle explicit file path
             if not os.path.isabs(pattern):
                 pattern = os.path.join(data_dir, f"{pattern}.mktdata.ohlcv.db")
-            resolved_files.append(pattern)
+            resolved_files.append((date_pattern, pattern))
     return sorted(list(set(resolved_files)))  # Remove duplicates and sort
 
 
@@ -61,8 +68,7 @@ def get_instruments(args: argparse.Namespace, config: Dict) -> List[Dict[str, st
 
 def run_backtest(
     config: Dict,
-    datafile: str,
-    price_column: str,
+    datafiles: List[str],
     fit_method: PairsTradingFitMethod,
     instruments: List[Dict[str, str]],
 ) -> BacktestResult:
@@ -70,12 +76,19 @@ def run_backtest(
     Run backtest for all pairs using the specified instruments.
     """
     bt_result: BacktestResult = BacktestResult(config=config)
+    # if len(datafiles) < 2:
+    #     print(f"WARNING: insufficient data files: {datafiles}")
+    #     return bt_result
+
+    if not all([os.path.exists(datafile) for datafile in datafiles]):
+        print(f"WARNING: data file {datafiles} does not exist")
+        return bt_result
 
     pairs_trades = []
+
     pairs = create_pairs(
-        datafile=datafile,
+        datafiles=datafiles,
         fit_method=fit_method,
-        price_column=price_column,
         config=config,
         instruments=instruments,
     )
@@ -92,7 +105,6 @@ def run_backtest(
     bt_result.collect_single_day_results(pairs_trades)
     return bt_result
 
-
 def main() -> None:
     parser = argparse.ArgumentParser(description="Run pairs trading backtest.")
     parser.add_argument(
@@ -122,20 +134,13 @@ def main() -> None:
     config: Dict = load_config(args.config)
 
     # Dynamically instantiate fit method class
-    fit_method_class_name = config.get("fit_method_class", None)
-    assert fit_method_class_name is not None
-    module_name, class_name = fit_method_class_name.rsplit(".", 1)
-    module = importlib.import_module(module_name)
-    fit_method = getattr(module, class_name)()
+    fit_method = PairsTradingFitMethod.create(config)
 
     # Resolve data files (CLI takes priority over config)
     instruments = get_instruments(args, config)
     datafiles = resolve_datafiles(config, args.date_pattern, instruments)
 
-    if not datafiles:
-        print("No data files found to process.")
-        return
-
+    days = list(set([day for day, _ in datafiles]))
     print(f"Found {len(datafiles)} data files to process:")
     for df in datafiles:
         print(f"  - {df}")
@@ -147,27 +152,26 @@ def main() -> None:
 
     # Initialize a dictionary to store all trade results
     all_results: Dict[str, Dict[str, Any]] = {}
-
-    # Store configuration in database for reference
-    if args.result_db.upper() != "NONE":
-        # Get list of all instruments for storage
-
-        # Remove duplicates while preserving order
-
-        store_config_in_database(
-            db_path=args.result_db,
-            config_file_path=args.config,
-            config=config,
-            fit_method_class=fit_method_class_name,
-            datafiles=datafiles,
-            instruments=instruments,
-        )
-
+    is_config_stored = False
     # Process each data file
-    price_column = config["price_column"]
 
-    for datafile in datafiles:
-        print(f"\n====== Processing {os.path.basename(datafile)} ======")
+    for day in sorted(days):
+        md_datafiles = [datafile for md_day, datafile in datafiles if md_day == day]
+        if not all([os.path.exists(datafile) for datafile in md_datafiles]):
+            print(f"WARNING: insufficient data files: {md_datafiles}")
+            continue
+        print(f"\n====== Processing {day} ======")
+
+        if not is_config_stored:    
+            store_config_in_database(
+                db_path=args.result_db,
+                config_file_path=args.config,
+                config=config,
+                fit_method_class=config["fit_method_class"],
+                datafiles=datafiles,
+                instruments=instruments,
+            )
+            is_config_stored = True
 
         # Process data for this file
         try:
@@ -175,14 +179,17 @@ def main() -> None:
 
             bt_results = run_backtest(
                 config=config,
-                datafile=datafile,
-                price_column=price_column,
+                datafiles=md_datafiles,
                 fit_method=fit_method,
                 instruments=instruments,
             )
+            
+            if bt_results.trades is None or len(bt_results.trades) == 0:
+                print(f"No trades found for {day}")
+                continue
 
-            # Store results with file name as key
-            filename = os.path.basename(datafile)
+            # Store results with day name as key
+            filename = os.path.basename(day)
             all_results[filename] = {
                 "trades": bt_results.trades.copy(),
                 "outstanding_positions": bt_results.outstanding_positions.copy(),
@@ -198,12 +205,12 @@ def main() -> None:
                         }
                     }
                 )
-                bt_results.store_results_in_database(args.result_db, datafile)
+                bt_results.store_results_in_database(db_path=args.result_db, day=day)
 
             print(f"Successfully processed {filename}")
 
         except Exception as err:
-            print(f"Error processing {datafile}: {str(err)}")
+            print(f"Error processing {day}: {str(err)}")
             import traceback
 
             traceback.print_exc()

research/research_tools.py

@@ -2,9 +2,9 @@ import glob
 import os
 from typing import Dict, List, Optional
 
+import pandas as pd
 from pt_trading.fit_method import PairsTradingFitMethod
 
-
 def resolve_datafiles(config: Dict, cli_datafiles: Optional[str] = None) -> List[str]:
     """
     Resolve the list of data files to process.
@@ -45,14 +45,13 @@ def resolve_datafiles(config: Dict, cli_datafiles: Optional[str] = None) -> List
 
 
 def create_pairs(
-    datafile: str,
+    datafiles: List[str],
     fit_method: PairsTradingFitMethod,
-    price_column: str,
     config: Dict,
     instruments: List[Dict[str, str]],
 ) -> List:
-    from tools.data_loader import load_market_data
     from pt_trading.trading_pair import TradingPair
+    from tools.data_loader import load_market_data
 
     all_indexes = range(len(instruments))
     unique_index_pairs = [(i, j) for i in all_indexes for j in all_indexes if i < j]
@@ -61,23 +60,34 @@ def create_pairs(
     # Update config to use the specified instruments
     config_copy = config.copy()
     config_copy["instruments"] = instruments
-    
-    market_data_df = load_market_data(
-        datafile=datafile,
-        instruments=instruments,
-        db_table_name=config_copy["market_data_loading"][instruments[0]["instrument_type"]]["db_table_name"],
-        trading_hours=config_copy["trading_hours"],
-    )
+
+    market_data_df = pd.DataFrame()
+    extra_minutes = 0
+    if "execution_price" in config_copy:
+        extra_minutes = config_copy["execution_price"]["shift"]
+
+    for datafile in datafiles:
+        md_df = load_market_data(
+            datafile = datafile,
+            instruments = instruments,
+            db_table_name = config_copy["market_data_loading"][instruments[0]["instrument_type"]]["db_table_name"],
+            trading_hours=config_copy["trading_hours"],
+            extra_minutes=extra_minutes,
+        )
+        market_data_df = pd.concat([market_data_df, md_df])
+
+    if len(set(market_data_df["symbol"])) != 2: # both symbols must be present for a pair
+        print(f"WARNING: insufficient data in files: {datafiles}")
+        return []
 
     for a_index, b_index in unique_index_pairs:
-        from research.pt_backtest import TradingPair
-
+        symbol_a=instruments[a_index]["symbol"]
+        symbol_b=instruments[b_index]["symbol"]
         pair = fit_method.create_trading_pair(
             config=config_copy,
             market_data=market_data_df,
-            symbol_a=instruments[a_index]["symbol"],
-            symbol_b=instruments[b_index]["symbol"],
-            price_column=price_column,
+            symbol_a=symbol_a,
+            symbol_b=symbol_b,
         )
         pairs.append(pair)
     return pairs

strategy/pair_strategy.py

@@ -23,7 +23,6 @@ from pt_trading.trading_pair import TradingPair
 def run_strategy(
     config: Dict,
     datafile: str,
-    price_column: str,
     fit_method: PairsTradingFitMethod,
     instruments: List[str],
 ) -> BacktestResult:
@@ -56,7 +55,6 @@ def run_strategy(
                 market_data=market_data_df,
                 symbol_a=instruments[a_index],
                 symbol_b=instruments[b_index],
-                price_column=price_column,
             )
             pairs.append(pair)
         return pairs
@@ -161,7 +159,6 @@ def main() -> None:
         )
 
     # Process each data file
-    price_column = config["price_column"]
 
     for datafile in datafiles:
         print(f"\n====== Processing {os.path.basename(datafile)} ======")
@@ -187,7 +184,6 @@ def main() -> None:
             bt_results = run_strategy(
                 config=config,
                 datafile=datafile,
-                price_column=price_column,
                 fit_method=fit_method,
                 instruments=instruments,
             )