progress

README.md

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

configuration/vecm.cfg

@@ -14,8 +14,11 @@
 
     # ====== Funding ======
     "funding_per_pair": 2000.0,
+  
     # ====== Trading Parameters ======
-    "price_column": "close",
+    # "stat_model_price": "close",
+    "stat_model_price": "vwap",
+  
     "execution_price": {
         "column": "vwap",
         "shift": 1,

configuration/zscore.cfg

@@ -15,7 +15,7 @@
     # ====== Funding ======
     "funding_per_pair": 2000.0,
     # ====== Trading Parameters ======
-    "price_column": "close",
+    "stat_model_price": "close",
     "execution_price": {
         "column": "vwap",
         "shift": 1,

lib/pt_trading/fit_method.py

@@ -43,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

@@ -431,7 +431,7 @@ class BacktestResult:
                         f" Close Dis-eq: {trd['open_scaled_disequilibrium']:.2f}"
 
                     print(
-                        f" {trd['open_time'].time()}  {trd['symbol']}: "
+                        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}"
@@ -552,7 +552,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]
 

lib/pt_trading/rolling_window_fit.py

@@ -146,8 +146,13 @@ class RollingFit(PairsTradingFitMethod):
             print(f"{pair}:  *** Position is NOT CLOSED. ***")
             # outstanding positions
             if config["close_outstanding_positions"]:
+                close_position_row = pd.Series(pair.market_data_.iloc[-2])
+                close_position_row["disequilibrium"] = 0.0
+                close_position_row["scaled_disequilibrium"] = 0.0
+                close_position_row["signed_scaled_disequilibrium"] = 0.0
+                
                 close_position_trades = self._get_close_trades(
-                    pair=pair, row=pred_row, close_threshold=close_threshold
+                    pair=pair, row=close_position_row, close_threshold=close_threshold
                 )
                 if close_position_trades is not None:
                     close_position_trades["status"] = PairState.CLOSE_POSITION.name
@@ -171,9 +176,10 @@ class RollingFit(PairsTradingFitMethod):
     def _get_open_trades(
         self, pair: TradingPair, row: pd.Series, open_threshold: float
     ) -> Optional[pd.DataFrame]:
-        colname_a, colname_b = pair.colnames()
+        colname_a, colname_b = pair.exec_prices_colnames()
 
         open_row = row
+
         open_tstamp = open_row["tstamp"]
         open_disequilibrium = open_row["disequilibrium"]
         open_scaled_disequilibrium = open_row["scaled_disequilibrium"]
@@ -252,7 +258,7 @@ class RollingFit(PairsTradingFitMethod):
     def _get_close_trades(
         self, pair: TradingPair, row: pd.Series, close_threshold: float
     ) -> Optional[pd.DataFrame]:
-        colname_a, colname_b = pair.colnames()
+        colname_a, colname_b = pair.exec_prices_colnames()
 
         close_row = row
         close_tstamp = close_row["tstamp"]

lib/pt_trading/trading_pair.py

@@ -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
@@ -91,17 +91,17 @@ class TradingPair(ABC):
         market_data: pd.DataFrame,
         symbol_a: str,
         symbol_b: str,
-        price_column: str,
     ):
         self.symbol_a_ = symbol_a
         self.symbol_b_ = symbol_b
-        self.price_column_ = price_column
+        self.stat_model_price_ = config["stat_model_price"]
-        self.set_market_data(market_data)
         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()]
         )
@@ -109,6 +109,22 @@ class TradingPair(ABC):
         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._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_:
@@ -139,7 +155,7 @@ class TradingPair(ABC):
     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
@@ -157,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_],
                 }
             )
 
@@ -201,8 +217,14 @@ class TradingPair(ABC):
 
     def colnames(self) -> List[str]:
         return [
-            f"{self.price_column_}_{self.symbol_a_}",
+            f"{self.stat_model_price_}_{self.symbol_a_}",
-            f"{self.price_column_}_{self.symbol_b_}",
+            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:
@@ -331,7 +353,7 @@ class TradingPair(ABC):
                 instrument_open_price = instrument_open_trades["price"].iloc[0]
 
                 sign = -1 if instrument_open_trades["side"].iloc[0] == "SELL" else 1
-                instrument_price = predicted_row[f"{self.price_column_}_{symbol}"]
+                instrument_price = predicted_row[f"{self.stat_model_price_}_{symbol}"]
                 instrument_return = (
                     sign
                     * (instrument_price - instrument_open_price)

lib/pt_trading/vecm_rolling_fit.py

@@ -7,15 +7,23 @@ from pt_trading.trading_pair import TradingPair
 from statsmodels.tsa.vector_ar.vecm import VECM, VECMResults
 
 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):
+    def __init__(
-        super().__init__(config, market_data, symbol_a, symbol_b, price_column)
+        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
-        
+
     def _train_pair(self) -> None:
         self._fit_VECM()
         assert self.vecm_fit_ is not None
@@ -51,7 +59,7 @@ class VECMTradingPair(TradingPair):
 
     def predict(self) -> pd.DataFrame:
         self._train_pair()
-        
+
         assert self.testing_df_ is not None
         assert self.vecm_fit_ is not None
         predicted_prices = self.vecm_fit_.predict(steps=len(self.testing_df_))
@@ -79,31 +87,36 @@ class VECMTradingPair(TradingPair):
             predicted_df["disequilibrium"] - self.training_mu_
         ) / self.training_std_
 
-        predicted_df["scaled_disequilibrium"] = (
+        predicted_df["scaled_disequilibrium"] = abs(
-            abs(predicted_df["signed_scaled_disequilibrium"])
+            predicted_df["signed_scaled_disequilibrium"]
         )
-        
+
         predicted_df = predicted_df.reset_index(drop=True)
         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)
+            self.pair_predict_result_ = pd.concat(
-        # Reset index to ensure proper indexing 
+                [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)
         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
         )

lib/pt_trading/z-score_rolling_fit.py

@@ -7,24 +7,34 @@ from pt_trading.trading_pair import TradingPair
 import statsmodels.api as sm
 
 NanoPerMin = 1e9
+
+
 class ZScoreTradingPair(TradingPair):
     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):
+    def __init__(
-        super().__init__(config, market_data, symbol_a, symbol_b, price_column)
+        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)
+        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
@@ -32,14 +42,14 @@ class ZScoreTradingPair(TradingPair):
 
         # Calculate spread and Z-score
         spread = symbol_a_px_series - hedge_ratio * symbol_b_px_series
-        self.zscore_df_  = (spread - spread.mean()) / spread.std()        
+        self.zscore_df_ = (spread - spread.mean()) / spread.std()
 
     def predict(self) -> pd.DataFrame:
         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_
@@ -47,28 +57,29 @@ class ZScoreTradingPair(TradingPair):
         predicted_df["disequilibrium"] = self.zscore_df_
         predicted_df["signed_scaled_disequilibrium"] = self.zscore_df_
         predicted_df["scaled_disequilibrium"] = abs(self.zscore_df_)
-        
+
         predicted_df = predicted_df.reset_index(drop=True)
         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)
+            self.pair_predict_result_ = pd.concat(
-        # Reset index to ensure proper indexing 
+                [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)
         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
         )

lib/tools/data_loader.py

@@ -28,13 +28,14 @@ def load_sqlite_to_dataframe(db_path:str, query:str) -> pd.DataFrame:
             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"))
@@ -85,7 +86,7 @@ def load_market_data(
             f"{date_str} {trading_hours['begin_session']}", trading_hours["timezone"]
         )
         end_time = convert_time_to_UTC(
-            f"{date_str} {trading_hours['end_session']}", trading_hours["timezone"]
+            f"{date_str} {trading_hours['end_session']}", trading_hours["timezone"], extra_minutes=2 # to get execution price
         )
 
         # Perform boolean selection

research/cointegration_test.py

@@ -85,7 +85,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 +105,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/pt_backtest.py

@@ -69,7 +69,6 @@ def get_instruments(args: argparse.Namespace, config: Dict) -> List[Dict[str, st
 def run_backtest(
     config: Dict,
     datafiles: List[str],
-    price_column: str,
     fit_method: PairsTradingFitMethod,
     instruments: List[Dict[str, str]],
 ) -> BacktestResult:
@@ -90,7 +89,6 @@ def run_backtest(
     pairs = create_pairs(
         datafiles=datafiles,
         fit_method=fit_method,
-        price_column=price_column,
         config=config,
         instruments=instruments,
     )
@@ -156,7 +154,6 @@ def main() -> None:
     all_results: Dict[str, Dict[str, Any]] = {}
     is_config_stored = False
     # Process each data file
-    price_column = config["price_column"]
 
     for day in sorted(days):
         md_datafiles = [datafile for md_day, datafile in datafiles if md_day == day]
@@ -183,7 +180,6 @@ def main() -> None:
             bt_results = run_backtest(
                 config=config,
                 datafiles=md_datafiles,
-                price_column=price_column,
                 fit_method=fit_method,
                 instruments=instruments,
             )

research/research_tools.py

@@ -48,7 +48,6 @@ def resolve_datafiles(config: Dict, cli_datafiles: Optional[str] = None) -> List
 def create_pairs(
     datafiles: List[str],
     fit_method: PairsTradingFitMethod,
-    price_column: str,
     config: Dict,
     instruments: List[Dict[str, str]],
 ) -> List:
@@ -85,7 +84,6 @@ def create_pairs(
             market_data=market_data_df,
             symbol_a=symbol_a,
             symbol_b=symbol_b,
-            price_column=price_column,
         )
         pairs.append(pair)
     return pairs

strategy/pair_strategy.py

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