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

@@ -38,13 +38,12 @@ CONFIG = EQT_CONFIG      # For equity data
 ```
 
 Each configuration dictionary specifies:
-- `security_type`: "CRYPTO" or "EQUITY".
 - `data_directory`: Path to the data files.
 - `datafiles`: A list of database files to process. You can comment/uncomment specific files to include/exclude them from the backtest.
 - `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

@@ -0,0 +1,42 @@
+{
+    "market_data_loading": {
+        "CRYPTO": {
+            "data_directory": "./data/crypto",
+            "db_table_name": "md_1min_bars",
+            "instrument_id_pfx": "PAIR-",
+        },
+        "EQUITY": {
+            "data_directory": "./data/equity",
+            "db_table_name": "md_1min_bars",
+            "instrument_id_pfx": "STOCK-",
+        }
+    },
+
+    # ====== Funding ======
+    "funding_per_pair": 2000.0,
+    # ====== Trading Parameters ======
+    "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,
+    "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": {
+        "timezone": "America/New_York",
+        "begin_session": "9:30:00",
+        "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/__DELETE__/static_fit.py

@@ -1,212 +0,0 @@
-from abc import ABC, abstractmethod
-from enum import Enum
-from typing import Dict, Optional, cast
-
-import pandas as pd  # type: ignore[import]
-from pt_trading.results import BacktestResult
-from pt_trading.trading_pair import TradingPair
-from pt_trading.fit_method import PairsTradingFitMethod
-
-NanoPerMin = 1e9
-
-
-
-class StaticFit(PairsTradingFitMethod):
-
-    def run_pair(
-        self, pair: TradingPair, bt_result: BacktestResult
-    ) -> Optional[pd.DataFrame]:  # abstractmethod
-        config = pair.config_
-        pair.get_datasets(training_minutes=config["training_minutes"])
-
-        try:
-            pair.predict()
-        except Exception as e:
-            print(f"{pair}: Prediction failed: {str(e)}")
-            return None
-
-        pair_trades = self.create_trading_signals(
-            pair=pair, config=config, result=bt_result
-        )
-
-        return pair_trades
-
-    def create_trading_signals(
-        self, pair: TradingPair, config: Dict, result: BacktestResult
-    ) -> pd.DataFrame:
-        beta = pair.vecm_fit_.beta  # type: ignore
-        colname_a, colname_b = pair.colnames()
-
-        predicted_df = pair.predicted_df_
-        if predicted_df is None:
-            # Return empty DataFrame with correct columns and dtypes
-            return pd.DataFrame(columns=self.TRADES_COLUMNS).astype({
-                "time": "datetime64[ns]",
-                "action": "string",
-                "symbol": "string", 
-                "price": "float64",
-                "disequilibrium": "float64",
-                "scaled_disequilibrium": "float64",
-                "pair": "object"
-            })
-
-        open_threshold = config["dis-equilibrium_open_trshld"]
-        close_threshold = config["dis-equilibrium_close_trshld"]
-
-        # Iterate through the testing dataset to find the first trading opportunity
-        open_row_index = None
-        for row_idx in range(len(predicted_df)):
-            curr_disequilibrium = predicted_df["scaled_disequilibrium"][row_idx]
-
-            # Check if current row has sufficient disequilibrium (not near-zero)
-            if curr_disequilibrium >= open_threshold:
-                open_row_index = row_idx
-                break
-
-        # If no row with sufficient disequilibrium found, skip this pair
-        if open_row_index is None:
-            print(f"{pair}: Insufficient disequilibrium in testing dataset. Skipping.")
-            return pd.DataFrame()
-
-        # Look for close signal starting from the open position
-        trading_signals_df = (
-            predicted_df["scaled_disequilibrium"][open_row_index:] < close_threshold
-        )
-
-        # Adjust indices to account for the offset from open_row_index
-        close_row_index = None
-        for idx, value in trading_signals_df.items():
-            if value:
-                close_row_index = idx
-                break
-
-        open_row = predicted_df.loc[open_row_index]
-        open_px_a = predicted_df.at[open_row_index, f"{colname_a}"]
-        open_px_b = predicted_df.at[open_row_index, f"{colname_b}"]
-        open_tstamp = predicted_df.at[open_row_index, "tstamp"]
-        open_disequilibrium = open_row["disequilibrium"]
-        open_scaled_disequilibrium = open_row["scaled_disequilibrium"]
-
-        abs_beta = abs(beta[1])
-        pred_px_b = predicted_df.loc[open_row_index][f"{colname_b}_pred"]
-        pred_px_a = predicted_df.loc[open_row_index][f"{colname_a}_pred"]
-
-        if pred_px_b * abs_beta - pred_px_a > 0:
-            open_side_a = "BUY"
-            open_side_b = "SELL"
-            close_side_a = "SELL"
-            close_side_b = "BUY"
-        else:
-            open_side_b = "BUY"
-            open_side_a = "SELL"
-            close_side_b = "SELL"
-            close_side_a = "BUY"
-
-        # If no close signal found, print position and unrealized PnL
-        if close_row_index is None:
-
-            last_row_index = len(predicted_df) - 1
-
-            # Use the new method from BacktestResult to handle outstanding positions
-            result.handle_outstanding_position(
-                pair=pair,
-                pair_result_df=predicted_df,
-                last_row_index=last_row_index,
-                open_side_a=open_side_a,
-                open_side_b=open_side_b,
-                open_px_a=float(open_px_a),
-                open_px_b=float(open_px_b),
-                open_tstamp=pd.Timestamp(open_tstamp),
-            )
-
-            # Return only open trades (no close trades)
-            trd_signal_tuples = [
-                (
-                    open_tstamp,
-                    open_side_a,
-                    pair.symbol_a_,
-                    open_px_a,
-                    open_disequilibrium,
-                    open_scaled_disequilibrium,
-                    pair,
-                ),
-                (
-                    open_tstamp,
-                    open_side_b,
-                    pair.symbol_b_,
-                    open_px_b,
-                    open_disequilibrium,
-                    open_scaled_disequilibrium,
-                    pair,
-                ),
-            ]
-        else:
-            # Close signal found - create complete trade
-            close_row = predicted_df.loc[close_row_index]
-            close_tstamp = close_row["tstamp"]
-            close_disequilibrium = close_row["disequilibrium"]
-            close_scaled_disequilibrium = close_row["scaled_disequilibrium"]
-            close_px_a = close_row[f"{colname_a}"]
-            close_px_b = close_row[f"{colname_b}"]
-
-            print(f"{pair}: Close signal found at index {close_row_index}")
-
-            trd_signal_tuples = [
-                (
-                    open_tstamp,
-                    open_side_a,
-                    pair.symbol_a_,
-                    open_px_a,
-                    open_disequilibrium,
-                    open_scaled_disequilibrium,
-                    pair,
-                ),
-                (
-                    open_tstamp,
-                    open_side_b,
-                    pair.symbol_b_,
-                    open_px_b,
-                    open_disequilibrium,
-                    open_scaled_disequilibrium,
-                    pair,
-                ),
-                (
-                    close_tstamp,
-                    close_side_a,
-                    pair.symbol_a_,
-                    close_px_a,
-                    close_disequilibrium,
-                    close_scaled_disequilibrium,
-                    pair,
-                ),
-                (
-                    close_tstamp,
-                    close_side_b,
-                    pair.symbol_b_,
-                    close_px_b,
-                    close_disequilibrium,
-                    close_scaled_disequilibrium,
-                    pair,
-                ),
-            ]
-
-        # Add tuples to data frame with explicit dtypes to avoid concatenation warnings
-        df = pd.DataFrame(
-            trd_signal_tuples,
-            columns=self.TRADES_COLUMNS,
-        )
-        # Ensure consistent dtypes
-        return df.astype({
-            "time": "datetime64[ns]",
-            "action": "string",
-            "symbol": "string", 
-            "price": "float64",
-            "disequilibrium": "float64",
-            "scaled_disequilibrium": "float64",
-            "pair": "object"
-        })
-
-    def reset(self) -> None:
-        pass
-
-

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
@@ -18,8 +20,18 @@ class PairsTradingFitMethod(ABC):
         "price",
         "disequilibrium",
         "scaled_disequilibrium",
+        "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(
@@ -31,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

@@ -46,7 +46,7 @@ def create_result_database(db_path: str) -> None:
         if db_dir and not os.path.exists(db_dir):
             os.makedirs(db_dir, exist_ok=True)
             print(f"Created directory: {db_dir}")
-        
+
         conn = sqlite3.connect(db_path)
         cursor = conn.cursor()
 
@@ -68,7 +68,8 @@ def create_result_database(db_path: str) -> None:
                 close_quantity INTEGER,
                 close_disequilibrium REAL,
                 symbol_return REAL,
-                pair_return REAL
+                pair_return REAL,
+                close_condition TEXT
             )
         """
         )
@@ -120,8 +121,8 @@ def store_config_in_database(
     config_file_path: str,
     config: Dict,
     fit_method_class: str,
-    datafiles: List[str],
-    instruments: List[str],
+    datafiles: List[Tuple[str, str]],
+    instruments: List[Dict[str, str]],
 ) -> None:
     """
     Store configuration information in the database for reference.
@@ -139,8 +140,13 @@ 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)
-        instruments_str = ", ".join(instruments)
+        datafiles_str = ", ".join([f"{datafile}" for _, datafile in datafiles])
+        instruments_str = ", ".join(
+            [
+                f"{inst['symbol']}:{inst['instrument_type']}:{inst['exchange_id']}"
+                for inst in instruments
+            ]
+        )
 
         # Insert configuration record
         cursor.execute(
@@ -170,6 +176,7 @@ def store_config_in_database(
 
         traceback.print_exc()
 
+
 def convert_timestamp(timestamp: Any) -> Optional[datetime]:
     """Convert pandas Timestamp to Python datetime object for SQLite compatibility."""
     if timestamp is None:
@@ -187,244 +194,6 @@ def convert_timestamp(timestamp: Any) -> Optional[datetime]:
     else:
         raise ValueError(f"Unsupported timestamp type: {type(timestamp)}")
 
- 
-def store_results_in_database(
-    db_path: str, datafile: str, bt_result: "BacktestResult"
-) -> None:
-    """
-    Store backtest results in the SQLite database.
-    """
-    if db_path.upper() == "NONE":
-        return
-
-    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
-
-        # Convert to proper date format
-        try:
-            date_obj = datetime.strptime(date_str, "%Y%m%d").date()
-        except ValueError:
-            # If date parsing fails, use current date
-            date_obj = datetime.now().date()
-
-        conn = sqlite3.connect(db_path)
-        cursor = conn.cursor()
-
-        # Process each trade from bt_result
-        trades = bt_result.get_trades()
-
-        for pair_name, symbols in trades.items():
-            # Calculate pair return for this pair
-            pair_return = 0.0
-            pair_trades = []
-
-            # First pass: collect all trades and calculate returns
-            for symbol, symbol_trades in symbols.items():
-                if len(symbol_trades) == 0:  # No trades for this symbol
-                    print(
-                        f"Warning: No trades found for symbol {symbol} in pair {pair_name}"
-                    )
-                    continue
-
-                elif len(symbol_trades) >= 2:  # Completed trades (entry + exit)
-                    # Handle both old and new tuple formats
-                    if len(symbol_trades[0]) == 2:  # Old format: (action, price)
-                        entry_action, entry_price = symbol_trades[0]
-                        exit_action, exit_price = symbol_trades[1]
-                        open_disequilibrium = 0.0  # Fallback for old format
-                        open_scaled_disequilibrium = 0.0
-                        close_disequilibrium = 0.0
-                        close_scaled_disequilibrium = 0.0
-                        open_time = datetime.now()
-                        close_time = datetime.now()
-                    else:  # New format: (action, price, disequilibrium, scaled_disequilibrium, timestamp)
-                        (
-                            entry_action,
-                            entry_price,
-                            open_disequilibrium,
-                            open_scaled_disequilibrium,
-                            open_time,
-                        ) = symbol_trades[0]
-                        (
-                            exit_action,
-                            exit_price,
-                            close_disequilibrium,
-                            close_scaled_disequilibrium,
-                            close_time,
-                        ) = symbol_trades[1]
-
-                        # Handle None values
-                        open_disequilibrium = (
-                            open_disequilibrium
-                            if open_disequilibrium is not None
-                            else 0.0
-                        )
-                        open_scaled_disequilibrium = (
-                            open_scaled_disequilibrium
-                            if open_scaled_disequilibrium is not None
-                            else 0.0
-                        )
-                        close_disequilibrium = (
-                            close_disequilibrium
-                            if close_disequilibrium is not None
-                            else 0.0
-                        )
-                        close_scaled_disequilibrium = (
-                            close_scaled_disequilibrium
-                            if close_scaled_disequilibrium is not None
-                            else 0.0
-                        )
-
-                        # Convert pandas Timestamps to Python datetime objects
-                        open_time = convert_timestamp(open_time) or datetime.now()
-                        close_time = convert_timestamp(close_time) or datetime.now()
-
-                    # Calculate actual share quantities based on funding per pair
-                    # Split funding equally between the two positions
-                    funding_per_position = bt_result.config["funding_per_pair"] / 2
-                    shares = funding_per_position / entry_price
-
-                    # Calculate symbol return
-                    symbol_return = 0.0
-                    if entry_action == "BUY" and exit_action == "SELL":
-                        symbol_return = (exit_price - entry_price) / entry_price * 100
-                    elif entry_action == "SELL" and exit_action == "BUY":
-                        symbol_return = (entry_price - exit_price) / entry_price * 100
-
-                    pair_return += symbol_return
-
-                    pair_trades.append(
-                        {
-                            "symbol": symbol,
-                            "entry_action": entry_action,
-                            "entry_price": entry_price,
-                            "exit_action": exit_action,
-                            "exit_price": exit_price,
-                            "symbol_return": symbol_return,
-                            "open_disequilibrium": open_disequilibrium,
-                            "open_scaled_disequilibrium": open_scaled_disequilibrium,
-                            "close_disequilibrium": close_disequilibrium,
-                            "close_scaled_disequilibrium": close_scaled_disequilibrium,
-                            "open_time": open_time,
-                            "close_time": close_time,
-                            "shares": shares,
-                            "is_completed": True,
-                        }
-                    )
-
-                # Skip one-sided trades - they will be handled by outstanding_positions table
-                elif len(symbol_trades) == 1:
-                    print(
-                        f"Skipping one-sided trade for {symbol} in pair {pair_name} - will be stored in outstanding_positions table"
-                    )
-                    continue
-
-                else:
-                    # This should not happen, but handle unexpected cases
-                    print(
-                        f"Warning: Unexpected number of trades ({len(symbol_trades)}) for symbol {symbol} in pair {pair_name}"
-                    )
-                    continue
-
-            # Second pass: insert completed trade records into database
-            for trade in pair_trades:
-                # Only store completed trades in pt_bt_results table
-                cursor.execute(
-                    """
-                    INSERT INTO pt_bt_results (
-                        date, pair, symbol, open_time, open_side, open_price, 
-                        open_quantity, open_disequilibrium, close_time, close_side, 
-                        close_price, close_quantity, close_disequilibrium, 
-                        symbol_return, pair_return
-                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-                """,
-                    (
-                        date_obj,
-                        pair_name,
-                        trade["symbol"],
-                        trade["open_time"],
-                        trade["entry_action"],
-                        trade["entry_price"],
-                        trade["shares"],
-                        trade["open_scaled_disequilibrium"],
-                        trade["close_time"],
-                        trade["exit_action"],
-                        trade["exit_price"],
-                        trade["shares"],
-                        trade["close_scaled_disequilibrium"],
-                        trade["symbol_return"],
-                        pair_return,
-                    ),
-                )
-
-        # Store outstanding positions in separate table
-        outstanding_positions = bt_result.get_outstanding_positions()
-        for pos in outstanding_positions:
-            # Calculate position quantity (negative for SELL positions)
-            position_qty_a = (
-                pos["shares_a"] if pos["side_a"] == "BUY" else -pos["shares_a"]
-            )
-            position_qty_b = (
-                pos["shares_b"] if pos["side_b"] == "BUY" else -pos["shares_b"]
-            )
-
-            # Calculate unrealized returns
-            # For symbol A: (current_price - open_price) / open_price * 100 * position_direction
-            unrealized_return_a = (
-                (pos["current_px_a"] - pos["open_px_a"]) / pos["open_px_a"] * 100
-            ) * (1 if pos["side_a"] == "BUY" else -1)
-            unrealized_return_b = (
-                (pos["current_px_b"] - pos["open_px_b"]) / pos["open_px_b"] * 100
-            ) * (1 if pos["side_b"] == "BUY" else -1)
-
-            # Store outstanding position for symbol A
-            cursor.execute(
-                """
-                INSERT INTO outstanding_positions (
-                    date, pair, symbol, position_quantity, last_price, unrealized_return, open_price, open_side
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-            """,
-                (
-                    date_obj,
-                    pos["pair"],
-                    pos["symbol_a"],
-                    position_qty_a,
-                    pos["current_px_a"],
-                    unrealized_return_a,
-                    pos["open_px_a"],
-                    pos["side_a"],
-                ),
-            )
-
-            # Store outstanding position for symbol B
-            cursor.execute(
-                """
-                INSERT INTO outstanding_positions (
-                    date, pair, symbol, position_quantity, last_price, unrealized_return, open_price, open_side
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
-            """,
-                (
-                    date_obj,
-                    pos["pair"],
-                    pos["symbol_b"],
-                    position_qty_b,
-                    pos["current_px_b"],
-                    unrealized_return_b,
-                    pos["open_px_b"],
-                    pos["side_b"],
-                ),
-            )
-
-        conn.commit()
-        conn.close()
-
-    except Exception as e:
-        print(f"Error storing results in database: {str(e)}")
-        import traceback
-
-        traceback.print_exc()
 
 
 class BacktestResult:
@@ -437,7 +206,8 @@ class BacktestResult:
         self.trades: Dict[str, Dict[str, Any]] = {}
         self.total_realized_pnl = 0.0
         self.outstanding_positions: List[Dict[str, Any]] = []
-
+        self.pairs_trades_: Dict[str, List[Dict[str, Any]]] = {}
+        
     def add_trade(
         self,
         pair_nm: str,
@@ -458,15 +228,16 @@ class BacktestResult:
         if symbol not in self.trades[pair_nm]:
             self.trades[pair_nm][symbol] = []
         self.trades[pair_nm][symbol].append(
-            {"symbol":symbol, 
-             "side":side, 
-             "action":action, 
-             "price":price, 
-             "disequilibrium":disequilibrium, 
-             "scaled_disequilibrium":scaled_disequilibrium, 
-             "timestamp":timestamp,
-             "status":status
-             }
+            {
+                "symbol": symbol,
+                "side": side,
+                "action": action,
+                "price": price,
+                "disequilibrium": disequilibrium,
+                "scaled_disequilibrium": scaled_disequilibrium,
+                "timestamp": timestamp,
+                "status": status,
+            }
         )
 
     def add_outstanding_position(self, position: Dict[str, Any]) -> None:
@@ -549,97 +320,128 @@ class BacktestResult:
 
     def calculate_returns(self, all_results: Dict[str, Dict[str, Any]]) -> None:
         """Calculate and print returns by day and pair."""
+        def _symbol_return(trade1_side: str, trade1_px: float, trade2_side: str, trade2_px: float) -> float:
+            if trade1_side == "BUY" and trade2_side == "SELL":
+                return (trade2_px - trade1_px) / trade1_px * 100
+            elif trade1_side == "SELL" and trade2_side == "BUY":
+                return (trade1_px - trade2_px) / trade1_px * 100
+            else:
+                return 0
+
         print("\n====== Returns By Day and Pair ======")
 
+        trades = []
         for filename, data in all_results.items():
-            day_return = 0
+            pairs = list(data["trades"].keys())
+            for pair in pairs:
+                self.pairs_trades_[pair] = []
+                trades_dict = data["trades"][pair]
+                for symbol in trades_dict.keys():
+                    trades.extend(trades_dict[symbol])
+            trades = sorted(trades, key=lambda x: (x["timestamp"], x["symbol"]))
+
             print(f"\n--- {filename} ---")
 
             self.outstanding_positions = data["outstanding_positions"]
+            
+            day_return = 0.0
+            for idx in range(0, len(trades), 4):
+                symbol_a = trades[idx]["symbol"]
+                trade_a_1 = trades[idx]
+                trade_a_2 = trades[idx + 2]
 
-            # Process each pair
-            for pair, symbols in data["trades"].items():
-                pair_return = 0
-                pair_trades = []
+                symbol_b = trades[idx + 1]["symbol"]
+                trade_b_1 = trades[idx + 1]
+                trade_b_2 = trades[idx + 3]
 
-                # Calculate individual symbol returns in the pair
-                for symbol, trades in symbols.items():
-                    if len(trades) == 0:
-                        continue
-                    symbol_return = 0
-                    symbol_trades =  [trade for trade in trades if trade["symbol"] == symbol]
+                symbol_return = 0
+                assert (
+                    trade_a_1["timestamp"] < trade_a_2["timestamp"]
+                ), f"Trade 1: {trade_a_1['timestamp']} is not less than Trade 2: {trade_a_2['timestamp']}"
+                assert (
+                    trade_a_1["action"] == "OPEN" and trade_a_2["action"] == "CLOSE"
+                ), f"Trade 1: {trade_a_1['action']} and Trade 2: {trade_a_2['action']} are the same"
 
-                    # Calculate returns for all trade combinations
-                    for idx in range(0, len(symbol_trades), 2):
-                        trade1 = trades[idx]
-                        trade2 = trades[idx + 1]
-                        
-                        assert trade1["timestamp"] < trade2["timestamp"], f"Trade 1: {trade1['timestamp']} is not less than Trade 2: {trade2['timestamp']}"
-                        assert trade1["action"] == "OPEN" and trade2["action"] == "CLOSE", f"Trade 1: {trade1['action']} and Trade 2: {trade2['action']} are the same"
-                        
-                        # Calculate return based on action combination
-                        trade_return = 0
-                        if trade1["side"] == "BUY" and trade2["side"] == "SELL":
-                            # Long position
-                            trade_return = (trade2["price"] - trade1["price"]) / trade1["price"] * 100
-                        elif trade1["side"] == "SELL" and trade2["side"] == "BUY":
-                            # Short position
-                            trade_return = (trade1["price"] - trade2["price"]) / trade1["price"] * 100
-                        
-                        symbol_return += trade_return
-                        
-                        # Store trade details for reporting
-                        pair_trades.append(
-                            (
-                                symbol,
-                                trade1["timestamp"],
-                                trade2["timestamp"],
-                                trade1["side"],
-                                trade1["price"],
-                                trade2["side"],
-                                trade2["price"],
-                                trade_return,
-                                trade1["scaled_disequilibrium"],
-                                trade2["scaled_disequilibrium"],
-                                f"{idx + 1}",  # Trade sequence number
-                            )
+                # Calculate return based on action combination
+                trade_return = 0
+                symbol_a_return = _symbol_return(trade_a_1["side"], trade_a_1["price"], trade_a_2["side"], trade_a_2["price"])
+                symbol_b_return = _symbol_return(trade_b_1["side"], trade_b_1["price"], trade_b_2["side"], trade_b_2["price"])
+
+                pair_return = symbol_a_return + symbol_b_return
+
+                self.pairs_trades_[pair].append(
+                    {
+                        "symbol": symbol_a,
+                        "open_side": trade_a_1["side"],
+                        "open_action": trade_a_1["action"],
+                        "open_price": trade_a_1["price"],
+                        "close_side": trade_a_2["side"],
+                        "close_action": trade_a_2["action"],
+                        "close_price": trade_a_2["price"],
+                        "symbol_return": symbol_a_return,
+                        "open_disequilibrium": trade_a_1["disequilibrium"],
+                        "open_scaled_disequilibrium": trade_a_1["scaled_disequilibrium"],
+                        "close_disequilibrium": trade_a_2["disequilibrium"],
+                        "close_scaled_disequilibrium": trade_a_2["scaled_disequilibrium"],
+                        "open_time": trade_a_1["timestamp"],
+                        "close_time": trade_a_2["timestamp"],
+                        "shares": self.config["funding_per_pair"] / 2  / trade_a_1["price"],
+                        "is_completed": True,
+                        "close_condition": trade_a_2["status"],
+                        "pair_return": pair_return
+                    }
+                )
+                self.pairs_trades_[pair].append(
+                    {
+                        "symbol": symbol_b,
+                        "open_side": trade_b_1["side"],
+                        "open_action": trade_b_1["action"],
+                        "open_price": trade_b_1["price"],
+                        "close_side": trade_b_2["side"],
+                        "close_action": trade_b_2["action"],
+                        "close_price": trade_b_2["price"],
+                        "symbol_return": symbol_b_return,
+                        "open_disequilibrium": trade_b_1["disequilibrium"],
+                        "open_scaled_disequilibrium": trade_b_1["scaled_disequilibrium"],
+                        "close_disequilibrium": trade_b_2["disequilibrium"],
+                        "close_scaled_disequilibrium": trade_b_2["scaled_disequilibrium"],
+                        "open_time": trade_b_1["timestamp"],
+                        "close_time": trade_b_2["timestamp"],
+                        "shares": self.config["funding_per_pair"] / 2  / trade_b_1["price"],
+                        "is_completed": True,
+                        "close_condition": trade_b_2["status"],
+                        "pair_return": pair_return
+                    }
+                )
+
+
+            # Print pair returns with disequilibrium information
+            day_return = 0.0
+            if pair in self.pairs_trades_:
+
+                print(f"{pair}:")
+                pair_return = 0.0
+                for trd in self.pairs_trades_[pair]:
+                    disequil_info = ""
+                    if (
+                        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["close_scaled_disequilibrium"]:.2f}'
                         )
-                    
-                    pair_return += symbol_return
 
-                # Print pair returns with disequilibrium information
-                if pair_trades:
-                    print(f"  {pair}:")
-                    for (
-                        symbol,
-                        trade1["timestamp"],
-                        trade2["timestamp"],
-                        trade1["side"],
-                        trade1["price"],
-                        trade2["side"],
-                        trade2["price"],
-                        trade_return,
-                        trade1["scaled_disequilibrium"],
-                        trade2["scaled_disequilibrium"],
-                        trade_num,
-                    ) in pair_trades:
-                        disequil_info = ""
-                        if (
-                            trade1["scaled_disequilibrium"] is not None
-                            and trade2["scaled_disequilibrium"] is not None
-                        ):
-                            disequil_info = f" | Open Dis-eq: {trade1["scaled_disequilibrium"]:.2f},"
-                            f" Close Dis-eq: {trade2["scaled_disequilibrium"]:.2f}"
+                    print(
+                        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"]
 
-                        print(
-                            f" {trade2['timestamp'].time()}  {symbol} (Trade #{trade_num}):"
-                            f" {trade1["side"]} @ ${trade1["price"]:.2f},"
-                            f" {trade2["side"]} @ ${trade2["price"]:.2f},"
-                            f" Return: {trade_return:.2f}%{disequil_info}"
-                        )
-                    print(f"    Pair Total Return: {pair_return:.2f}%")
-                    day_return += pair_return
-                    
+                print(f"    Pair Total Return: {pair_return:.2f}%")
+                day_return += pair_return
 
             # Print day total return and add to global realized PnL
             if day_return != 0:
@@ -716,7 +518,7 @@ class BacktestResult:
 
             print("-" * 100)
 
-            total_value += pos["total_current_value"] 
+            total_value += pos["total_current_value"]
 
         print(f"{'TOTAL OUTSTANDING VALUE':<80} ${total_value:<12.2f}")
 
@@ -752,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]
 
@@ -811,3 +613,131 @@ class BacktestResult:
         )
 
         return current_value_a, current_value_b, total_current_value
+    
+    def store_results_in_database(
+        self, db_path: str, day: str
+    ) -> None:
+        """
+        Store backtest results in the SQLite database.
+        """
+        if db_path.upper() == "NONE":
+            return
+
+        try:
+            # Extract date from datafile name (assuming format like 20250528.mktdata.ohlcv.db)
+            date_str = day
+
+            # Convert to proper date format
+            try:
+                date_obj = datetime.strptime(date_str, "%Y%m%d").date()
+            except ValueError:
+                # If date parsing fails, use current date
+                date_obj = datetime.now().date()
+
+            conn = sqlite3.connect(db_path)
+            cursor = conn.cursor()
+
+            # Process each trade from bt_result
+            trades = self.get_trades()
+
+            for pair_name, _ in trades.items():
+
+                # Second pass: insert completed trade records into database
+                for trade_pair in sorted(self.pairs_trades_[pair_name], key=lambda x: x["open_time"]):
+                    # Only store completed trades in pt_bt_results table
+                    cursor.execute(
+                        """
+                        INSERT INTO pt_bt_results (
+                            date, pair, symbol, open_time, open_side, open_price, 
+                            open_quantity, open_disequilibrium, close_time, close_side, 
+                            close_price, close_quantity, close_disequilibrium, 
+                            symbol_return, pair_return, close_condition
+                        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    """,
+                        (
+                            date_obj,
+                            pair_name,
+                            trade_pair["symbol"],
+                            trade_pair["open_time"],
+                            trade_pair["open_side"],
+                            trade_pair["open_price"],
+                            trade_pair["shares"],
+                            trade_pair["open_scaled_disequilibrium"],
+                            trade_pair["close_time"],
+                            trade_pair["close_side"],
+                            trade_pair["close_price"],
+                            trade_pair["shares"],
+                            trade_pair["close_scaled_disequilibrium"],
+                            trade_pair["symbol_return"],
+                            trade_pair["pair_return"],
+                            trade_pair["close_condition"]
+                        ),
+                    )
+
+            # Store outstanding positions in separate table
+            outstanding_positions = self.get_outstanding_positions()
+            for pos in outstanding_positions:
+                # Calculate position quantity (negative for SELL positions)
+                position_qty_a = (
+                    pos["shares_a"] if pos["side_a"] == "BUY" else -pos["shares_a"]
+                )
+                position_qty_b = (
+                    pos["shares_b"] if pos["side_b"] == "BUY" else -pos["shares_b"]
+                )
+
+                # Calculate unrealized returns
+                # For symbol A: (current_price - open_price) / open_price * 100 * position_direction
+                unrealized_return_a = (
+                    (pos["current_px_a"] - pos["open_px_a"]) / pos["open_px_a"] * 100
+                ) * (1 if pos["side_a"] == "BUY" else -1)
+                unrealized_return_b = (
+                    (pos["current_px_b"] - pos["open_px_b"]) / pos["open_px_b"] * 100
+                ) * (1 if pos["side_b"] == "BUY" else -1)
+
+                # Store outstanding position for symbol A
+                cursor.execute(
+                    """
+                    INSERT INTO outstanding_positions (
+                        date, pair, symbol, position_quantity, last_price, unrealized_return, open_price, open_side
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                    (
+                        date_obj,
+                        pos["pair"],
+                        pos["symbol_a"],
+                        position_qty_a,
+                        pos["current_px_a"],
+                        unrealized_return_a,
+                        pos["open_px_a"],
+                        pos["side_a"],
+                    ),
+                )
+
+                # Store outstanding position for symbol B
+                cursor.execute(
+                    """
+                    INSERT INTO outstanding_positions (
+                        date, pair, symbol, position_quantity, last_price, unrealized_return, open_price, open_side
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                    (
+                        date_obj,
+                        pos["pair"],
+                        pos["symbol_b"],
+                        position_qty_b,
+                        pos["current_px_b"],
+                        unrealized_return_b,
+                        pos["open_px_b"],
+                        pos["side_b"],
+                    ),
+                )
+
+            conn.commit()
+            conn.close()
+
+        except Exception as e:
+            print(f"Error storing results in database: {str(e)}")
+            import traceback
+
+            traceback.print_exc()
+

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,17 +176,20 @@ 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"]
+        signed_scaled_disequilibrium = open_row["signed_scaled_disequilibrium"]
         open_px_a = open_row[f"{colname_a}"]
         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"
@@ -214,6 +222,7 @@ class RollingFit(PairsTradingFitMethod):
                 open_px_a,
                 open_disequilibrium,
                 open_scaled_disequilibrium,
+                signed_scaled_disequilibrium,
                 pair,
             ),
             (
@@ -224,14 +233,12 @@ class RollingFit(PairsTradingFitMethod):
                 open_px_b,
                 open_disequilibrium,
                 open_scaled_disequilibrium,
+                signed_scaled_disequilibrium,
                 pair,
             ),
         ]
         # 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(
             {
@@ -241,6 +248,7 @@ class RollingFit(PairsTradingFitMethod):
                 "price": "float64",
                 "disequilibrium": "float64",
                 "scaled_disequilibrium": "float64",
+                "signed_scaled_disequilibrium": "float64",
                 "pair": "object",
             }
         )
@@ -248,12 +256,13 @@ 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"]
         close_disequilibrium = close_row["disequilibrium"]
         close_scaled_disequilibrium = close_row["scaled_disequilibrium"]
+        signed_scaled_disequilibrium = close_row["signed_scaled_disequilibrium"]
         close_px_a = close_row[f"{colname_a}"]
         close_px_b = close_row[f"{colname_b}"]
 
@@ -269,6 +278,7 @@ class RollingFit(PairsTradingFitMethod):
                 close_px_a,
                 close_disequilibrium,
                 close_scaled_disequilibrium,
+                signed_scaled_disequilibrium,
                 pair,
             ),
             (
@@ -279,6 +289,7 @@ class RollingFit(PairsTradingFitMethod):
                 close_px_b,
                 close_disequilibrium,
                 close_scaled_disequilibrium,
+                signed_scaled_disequilibrium,
                 pair,
             ),
         ]
@@ -297,6 +308,7 @@ class RollingFit(PairsTradingFitMethod):
                 "price": "float64",
                 "disequilibrium": "float64",
                 "scaled_disequilibrium": "float64",
+                "signed_scaled_disequilibrium": "float64",
                 "pair": "object",
             }
         )

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,24 +312,30 @@ 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
                 return True
         if "loss" in config["stop_close_conditions"]:
             if current_return <= config["stop_close_conditions"]["loss"]:
+                print(f"STOP LOSS: {current_return}")
                 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:
@@ -300,18 +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_sign = -1 if instrument_open_trades["action"].iloc[0]  == "SELL" else 1
-                instrument_price = predicted_row[f"{self.price_column_}_{symbol}"]
-                instrument_return = instrument_sign * (instrument_price - instrument_open_trades["price"].iloc[0]) / instrument_open_trades["price"].iloc[0]
+                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.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()
 
@@ -324,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,70 +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["scaled_disequilibrium"] = (
-            abs(predicted_df["disequilibrium"] - self.training_mu_)
-            / self.training_std_
+        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 = 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,73 +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["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

@@ -1,10 +1,17 @@
+from __future__ import annotations
+
 import sqlite3
 from typing import Dict, List, cast
 import pandas as pd
 
 
+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
 
-def load_sqlite_to_dataframe(db_path, query):
     try:
         conn = sqlite3.connect(db_path)
 
@@ -21,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"))
@@ -35,25 +43,28 @@ def convert_time_to_UTC(value: str, timezone: str) -> str:
     return result.strftime("%Y-%m-%d %H:%M:%S")
 
 
-def load_market_data(datafile: str, config: Dict) -> pd.DataFrame:
-    from tools.data_loader import load_sqlite_to_dataframe
+def load_market_data(
+    datafile: str,
+    instruments: List[Dict[str, str]],
+    db_table_name: str,
+    trading_hours: Dict = {},
+    extra_minutes: int = 0,
+) -> pd.DataFrame:
 
-    instrument_ids = [
-        '"' + config["instrument_id_pfx"] + instrument + '"'
-        for instrument in config["instruments"]
+    insts = [
+        '"' + instrument["instrument_id_pfx"] + instrument["symbol"] + '"'
+        for instrument in instruments
     ]
-    security_type = config["security_type"]
-    exchange_id = config["exchange_id"]
+    instrument_ids = list(set(insts))
+    exchange_ids = list(
+        set(['"' + instrument["exchange_id"] + '"' for instrument in instruments])
+    )
 
     query = "select"
-    if security_type == "CRYPTO":
-        query += " strftime('%Y-%m-%d %H:%M:%S', tstamp_ns/1000000000, 'unixepoch') as tstamp"
-        query += ", tstamp as time_ns"
-    else:
-        query += " tstamp"
-        query += ", tstamp_ns as time_ns"
+    query += " tstamp"
+    query += ", tstamp_ns as time_ns"
 
-    query += f", substr(instrument_id, {len(config['instrument_id_pfx']) + 1}) as symbol"
+    query += f", substr(instrument_id, instr(instrument_id, '-') + 1) as symbol"
     query += ", open"
     query += ", high"
     query += ", low"
@@ -62,74 +73,76 @@ def load_market_data(datafile: str, config: Dict) -> pd.DataFrame:
     query += ", num_trades"
     query += ", vwap"
 
-    query += f" from {config['db_table_name']}"
-    query += f" where exchange_id ='{exchange_id}'"
+    query += f" from {db_table_name}"
+    query += f" where exchange_id in ({','.join(exchange_ids)})"
     query += f" and instrument_id in ({','.join(instrument_ids)})"
 
     df = load_sqlite_to_dataframe(db_path=datafile, query=query)
 
     # Trading Hours
-    date_str = df["tstamp"][0][0:10]
-    trading_hours = config["trading_hours"]
+    if len(df) > 0 and len(trading_hours) > 0:
+        date_str = df["tstamp"][0][0:10]
 
-    start_time = convert_time_to_UTC(
-        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"]
-    )
+        start_time = convert_time_to_UTC(
+            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"], extra_minutes=extra_minutes # to get execution price
+        )
 
-    # Perform boolean selection
-    df = df[(df["tstamp"] >= start_time) & (df["tstamp"] <= end_time)]
-    df["tstamp"] = pd.to_datetime(df["tstamp"])
+        # Perform boolean selection
+        df = df[(df["tstamp"] >= start_time) & (df["tstamp"] <= end_time)]
+        df["tstamp"] = pd.to_datetime(df["tstamp"])
 
     return cast(pd.DataFrame, df)
 
 
-def get_available_instruments_from_db(datafile: str, config: Dict) -> List[str]:
-    """
-    Auto-detect available instruments from the database by querying distinct instrument_id values.
-    Returns instruments without the configured prefix.
-    """
-    try:
-        conn = sqlite3.connect(datafile)
+# def get_available_instruments_from_db(datafile: str, config: Dict) -> List[str]:
+#     """
+#     Auto-detect available instruments from the database by querying distinct instrument_id values.
+#     Returns instruments without the configured prefix.
+#     """
+#     try:
+#         conn = sqlite3.connect(datafile)
 
-        # Build exclusion list with full instrument_ids
-        exclude_instruments = config.get("exclude_instruments", [])
-        prefix = config.get("instrument_id_pfx", "")
-        exclude_instrument_ids = [f"{prefix}{inst}" for inst in exclude_instruments]
-        
-        # Query to get distinct instrument_ids
-        query = f"""
-        SELECT DISTINCT instrument_id 
-        FROM {config['db_table_name']} 
-        WHERE exchange_id = ?
-        """
-        
-        # Add exclusion clause if there are instruments to exclude
-        if exclude_instrument_ids:
-            placeholders = ','.join(['?' for _ in exclude_instrument_ids])
-            query += f" AND instrument_id NOT IN ({placeholders})"
-            cursor = conn.execute(query, (config["exchange_id"],) + tuple(exclude_instrument_ids))
-        else:
-            cursor = conn.execute(query, (config["exchange_id"],))
-        instrument_ids = [row[0] for row in cursor.fetchall()]
-        conn.close()
+#         # Build exclusion list with full instrument_ids
+#         exclude_instruments = config.get("exclude_instruments", [])
+#         prefix = config.get("instrument_id_pfx", "")
+#         exclude_instrument_ids = [f"{prefix}{inst}" for inst in exclude_instruments]
 
-        # Remove the configured prefix to get instrument symbols
-        instruments = []
-        for instrument_id in instrument_ids:
-            if instrument_id.startswith(prefix):
-                symbol = instrument_id[len(prefix) :]
-                instruments.append(symbol)
-            else:
-                instruments.append(instrument_id)
+#         # Query to get distinct instrument_ids
+#         query = f"""
+#         SELECT DISTINCT instrument_id
+#         FROM {config['db_table_name']}
+#         WHERE exchange_id = ?
+#         """
 
-        return sorted(instruments)
+#         # Add exclusion clause if there are instruments to exclude
+#         if exclude_instrument_ids:
+#             placeholders = ",".join(["?" for _ in exclude_instrument_ids])
+#             query += f" AND instrument_id NOT IN ({placeholders})"
+#             cursor = conn.execute(
+#                 query, (config["exchange_id"],) + tuple(exclude_instrument_ids)
+#             )
+#         else:
+#             cursor = conn.execute(query, (config["exchange_id"],))
+#         instrument_ids = [row[0] for row in cursor.fetchall()]
+#         conn.close()
 
-    except Exception as e:
-        print(f"Error auto-detecting instruments from {datafile}: {str(e)}")
-        return []
+#         # Remove the configured prefix to get instrument symbols
+#         instruments = []
+#         for instrument_id in instrument_ids:
+#             if instrument_id.startswith(prefix):
+#                 symbol = instrument_id[len(prefix) :]
+#                 instruments.append(symbol)
+#             else:
+#                 instruments.append(instrument_id)
+
+#         return sorted(instruments)
+
+#     except Exception as e:
+#         print(f"Error auto-detecting instruments from {datafile}: {str(e)}")
+#         return []
 
 
 # if __name__ == "__main__":

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

@@ -8,19 +8,20 @@ from typing import Any, Dict, List, Optional
 import pandas as pd
 
 from tools.config import expand_filename, load_config
-from tools.data_loader import get_available_instruments_from_db, load_market_data
+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,82 +3,96 @@ 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
 
 from research.research_tools import create_pairs
 from tools.config import expand_filename, load_config
-from tools.data_loader import get_available_instruments_from_db, load_market_data
 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
 
+DayT = str
+DataFileNameT = str
 
-def resolve_datafiles(config: Dict, cli_datafiles: Optional[str] = None) -> List[str]:
-    """
-    Resolve the list of data files to process.
-    CLI datafiles take priority over config datafiles.
-    Supports wildcards in config but not in CLI.
-    """
-    if cli_datafiles:
-        # CLI override - comma-separated list, no wildcards
-        datafiles = [f.strip() for f in cli_datafiles.split(",")]
-        # Make paths absolute relative to data directory
-        data_dir = config.get("data_directory", "./data")
-        resolved_files = []
-        for df in datafiles:
-            if not os.path.isabs(df):
-                df = os.path.join(data_dir, df)
-            resolved_files.append(df)
-        return resolved_files
-
-    # Use config datafiles with wildcard support
-    config_datafiles = config.get("datafiles", [])
-    data_dir = config.get("data_directory", "./data")
-    resolved_files = []
-
-    for pattern in config_datafiles:
+def resolve_datafiles(
+    config: Dict, date_pattern: str, instruments: List[Dict[str, str]]
+) -> List[Tuple[DayT, DataFileNameT]]:
+    resolved_files: List[Tuple[DayT, DataFileNameT]] = []
+    for inst in instruments:
+        pattern = date_pattern
+        inst_type = inst["instrument_type"]
+        data_dir = config["market_data_loading"][inst_type]["data_directory"]
         if "*" in pattern or "?" in pattern:
             # Handle wildcards
             if not os.path.isabs(pattern):
-                pattern = os.path.join(data_dir, 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, pattern)
-            resolved_files.append(pattern)
-
+                pattern = os.path.join(data_dir, f"{pattern}.mktdata.ohlcv.db")
+            resolved_files.append((date_pattern, pattern))
     return sorted(list(set(resolved_files)))  # Remove duplicates and sort
 
 
+def get_instruments(args: argparse.Namespace, config: Dict) -> List[Dict[str, str]]:
+
+    instruments = [
+        {
+            "symbol": inst.split(":")[0],
+            "instrument_type": inst.split(":")[1],
+            "exchange_id": inst.split(":")[2],
+            "instrument_id_pfx": config["market_data_loading"][inst.split(":")[1]][
+                "instrument_id_pfx"
+            ],
+            "db_table_name": config["market_data_loading"][inst.split(":")[1]][
+                "db_table_name"
+            ],
+        }
+        for inst in args.instruments.split(",")
+    ]
+    return instruments
+
+
 def run_backtest(
     config: Dict,
-    datafile: str,
-    price_column: str,
+    datafiles: List[str],
     fit_method: PairsTradingFitMethod,
-    instruments: List[str],
+    instruments: List[Dict[str, str]],
 ) -> BacktestResult:
     """
     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 = []
-    for pair in create_pairs(
-        datafile=datafile,
+
+    pairs = create_pairs(
+        datafiles=datafiles,
         fit_method=fit_method,
-        price_column=price_column,
         config=config,
         instruments=instruments,
-    ):
+    )
+    for pair in pairs:
         single_pair_trades = fit_method.run_pair(pair=pair, bt_result=bt_result)
         if single_pair_trades is not None and len(single_pair_trades) > 0:
             pairs_trades.append(single_pair_trades)
@@ -91,23 +105,22 @@ 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(
         "--config", type=str, required=True, help="Path to the configuration file."
     )
     parser.add_argument(
-        "--datafiles",
+        "--date_pattern",
         type=str,
-        required=False,
-        help="Comma-separated list of data files (overrides config). No wildcards supported.",
+        required=True,
+        help="Date YYYYMMDD, allows * and ? wildcards",
     )
     parser.add_argument(
         "--instruments",
         type=str,
-        required=False,
-        help="Comma-separated list of instrument symbols (e.g., COIN,GBTC). If not provided, auto-detects from database.",
+        required=True,
+        help="Comma-separated list of instrument symbols (e.g., COIN:EQUITY,GBTC:CRYPTO)",
     )
     parser.add_argument(
         "--result_db",
@@ -121,19 +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)
-    datafiles = resolve_datafiles(config, args.datafiles)
-
-    if not datafiles:
-        print("No data files found to process.")
-        return
+    instruments = get_instruments(args, config)
+    datafiles = resolve_datafiles(config, args.date_pattern, instruments)
 
+    days = list(set([day for day, _ in datafiles]))
     print(f"Found {len(datafiles)} data files to process:")
     for df in datafiles:
         print(f"  - {df}")
@@ -145,51 +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
-        all_instruments = []
-        for datafile in datafiles:
-            if args.instruments:
-                file_instruments = [
-                    inst.strip() for inst in args.instruments.split(",")
-                ]
-            else:
-                file_instruments = get_available_instruments_from_db(datafile, config)
-            all_instruments.extend(file_instruments)
-
-        # Remove duplicates while preserving order
-        unique_instruments = list(dict.fromkeys(all_instruments))
-
-        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=unique_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)} ======")
-
-        # Determine instruments to use
-        if args.instruments:
-            # Use CLI-specified instruments
-            instruments = [inst.strip() for inst in args.instruments.split(",")]
-            print(f"Using CLI-specified instruments: {instruments}")
-        else:
-            # Auto-detect instruments from database
-            instruments = get_available_instruments_from_db(datafile, config)
-            print(f"Auto-detected instruments: {instruments}")
-
-        if not instruments:
-            print(f"No instruments found for {datafile}, skipping...")
+    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:
@@ -197,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(),
@@ -212,12 +197,20 @@ def main() -> None:
 
             # Store results in database
             if args.result_db.upper() != "NONE":
-                store_results_in_database(args.result_db, datafile, bt_results)
+                bt_results.calculate_returns(
+                    {
+                        filename: {
+                            "trades": bt_results.trades.copy(),
+                            "outstanding_positions": bt_results.outstanding_positions.copy(),
+                        }
+                    }
+                )
+                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,10 +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.
@@ -44,28 +43,51 @@ def resolve_datafiles(config: Dict, cli_datafiles: Optional[str] = None) -> List
 
     return sorted(list(set(resolved_files)))  # Remove duplicates and sort
 
-def create_pairs(datafile: str, fit_method: PairsTradingFitMethod, price_column: str, config: Dict, instruments: List[str]) -> List:
-        from tools.data_loader import load_market_data
-        from pt_trading.trading_pair import TradingPair
-        all_indexes = range(len(instruments))
-        unique_index_pairs = [(i, j) for i in all_indexes for j in all_indexes if i < j]
-        pairs = []
 
-        # Update config to use the specified instruments
-        config_copy = config.copy()
-        config_copy["instruments"] = instruments
+def create_pairs(
+    datafiles: List[str],
+    fit_method: PairsTradingFitMethod,
+    config: Dict,
+    instruments: List[Dict[str, str]],
+) -> List:
+    from pt_trading.trading_pair import TradingPair
+    from tools.data_loader import load_market_data
 
-        market_data_df = load_market_data(datafile, config=config_copy)
+    all_indexes = range(len(instruments))
+    unique_index_pairs = [(i, j) for i in all_indexes for j in all_indexes if i < j]
+    pairs = []
 
-        for a_index, b_index in unique_index_pairs:
-            from research.pt_backtest import TradingPair
-            pair = fit_method.create_trading_pair(
-                config=config_copy,
-                market_data=market_data_df,
-                symbol_a=instruments[a_index],
-                symbol_b=instruments[b_index],
-                price_column=price_column,
-            )
-            pairs.append(pair)
-        return pairs
+    # Update config to use the specified instruments
+    config_copy = config.copy()
+    config_copy["instruments"] = instruments
 
+    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:
+        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=symbol_a,
+            symbol_b=symbol_b,
+        )
+        pairs.append(pair)
+    return pairs

strategy/pair_strategy.py

@@ -20,12 +20,9 @@ from pt_trading.fit_methods import PairsTradingFitMethod
 from pt_trading.trading_pair import TradingPair
 
 
-
-
 def run_strategy(
     config: Dict,
     datafile: str,
-    price_column: str,
     fit_method: PairsTradingFitMethod,
     instruments: List[str],
 ) -> BacktestResult:
@@ -44,14 +41,20 @@ def run_strategy(
         config_copy = config.copy()
         config_copy["instruments"] = instruments
 
-        market_data_df = load_market_data(datafile, config=config_copy)
+        market_data_df = load_market_data(
+            datafile=datafile,
+            exchange_id=config_copy["exchange_id"],
+            instruments=config_copy["instruments"],
+            instrument_id_pfx=config_copy["instrument_id_pfx"],
+            db_table_name=config_copy["db_table_name"],
+            trading_hours=config_copy["trading_hours"],
+        )
 
         for a_index, b_index in unique_index_pairs:
             pair = fit_method.create_trading_pair(
                 market_data=market_data_df,
                 symbol_a=instruments[a_index],
                 symbol_b=instruments[b_index],
-                price_column=price_column,
             )
             pairs.append(pair)
         return pairs
@@ -156,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)} ======")
@@ -182,7 +184,6 @@ def main() -> None:
             bt_results = run_strategy(
                 config=config,
                 datafile=datafile,
-                price_column=price_column,
                 fit_method=fit_method,
                 instruments=instruments,
             )