research/pairs_trading
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

progress

Browse Source
This commit is contained in:
Oleg Sheynin 2025-07-24 21:09:13 +00:00
parent a7b4777f76
commit af0a6f62a9
9 changed files with 247 additions and 6224 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
configuration/vecm.cfg Normal file
View File

@ -0,0 +1,38 @@
{
"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 ======
"price_column": "close",
"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": {
"timezone": "America/New_York",
"begin_session": "9:30:00",
"end_session": "18:30:00",
}
}

38
configuration/z-score-intermarket.cfg Normal file
View File

@ -0,0 +1,38 @@
{
"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 ======
"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": {
"timezone": "America/New_York",
"begin_session": "9:30:00",
"end_session": "15:30:00",
}
}

6
configuration/zscore.cfg
View File

@ -13,7 +13,7 @@
}, },
# ====== Funding ====== # ====== Funding ======
"funding_per_pair": 2000.0, "funding_per_pair": 2000.0,
# ====== Trading Parameters ====== # ====== Trading Parameters ======
"price_column": "close", "price_column": "close",
"dis-equilibrium_open_trshld": 2.0, "dis-equilibrium_open_trshld": 2.0,
@ -31,8 +31,8 @@
"close_outstanding_positions": true, "close_outstanding_positions": true,
# "close_outstanding_positions": false, # "close_outstanding_positions": false,
"trading_hours": { "trading_hours": {
"timezone": "America/New_York",
"begin_session": "9:30:00", "begin_session": "9:30:00",
"end_session": "22:30:00", "end_session": "18:30:00",
"timezone": "America/New_York"
} }
} }

9
lib/pt_trading/results.py
View File

@ -121,7 +121,7 @@ def store_config_in_database(
config_file_path: str, config_file_path: str,
config: Dict, config: Dict,
fit_method_class: str, fit_method_class: str,
datafiles: List[str], datafiles: List[Tuple[str, str]],
instruments: List[Dict[str, str]], instruments: List[Dict[str, str]],
) -> None: ) -> None:
""" """
@ -140,7 +140,7 @@ def store_config_in_database(
config_json = json.dumps(config, indent=2, default=str) config_json = json.dumps(config, indent=2, default=str)
# Convert lists to comma-separated strings for storage # Convert lists to comma-separated strings for storage
datafiles_str = ", ".join(datafiles) datafiles_str = ", ".join([f"{datafile}" for _, datafile in datafiles])
instruments_str = ", ".join( instruments_str = ", ".join(
[ [
f"{inst['symbol']}:{inst['instrument_type']}:{inst['exchange_id']}" f"{inst['symbol']}:{inst['instrument_type']}:{inst['exchange_id']}"
@ -613,7 +613,7 @@ class BacktestResult:
return current_value_a, current_value_b, total_current_value return current_value_a, current_value_b, total_current_value
def store_results_in_database( def store_results_in_database(
self, db_path: str, datafile: str self, db_path: str, day: str
) -> None: ) -> None:
""" """
Store backtest results in the SQLite database. Store backtest results in the SQLite database.
@ -623,8 +623,7 @@ class BacktestResult:
try: try:
# Extract date from datafile name (assuming format like 20250528.mktdata.ohlcv.db) # Extract date from datafile name (assuming format like 20250528.mktdata.ohlcv.db)
filename = os.path.basename(datafile) date_str = day
date_str = filename.split(".")[0] # Extract date part
# Convert to proper date format # Convert to proper date format
try: try:

103
lib/pt_trading/trading_pair.py
View File

@ -17,7 +17,7 @@ class PairState(Enum):
class CointegrationData: class CointegrationData:
EG_PVALUE_THRESHOLD = 0.05 EG_PVALUE_THRESHOLD = 0.05
tstamp_: pd.Timestamp tstamp_: pd.Timestamp
pair_: str pair_: str
eg_pvalue_: float eg_pvalue_: float
@ -63,7 +63,7 @@ class CointegrationData:
"johansen_cvt": self.johansen_cvt_, "johansen_cvt": self.johansen_cvt_,
"eg_is_cointegrated": self.eg_is_cointegrated_, "eg_is_cointegrated": self.eg_is_cointegrated_,
"johansen_is_cointegrated": self.johansen_is_cointegrated_, "johansen_is_cointegrated": self.johansen_is_cointegrated_,
} }
def __repr__(self) -> str: 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_})" 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_})"
@ -86,7 +86,12 @@ class TradingPair(ABC):
# predicted_df_: Optional[pd.DataFrame] # predicted_df_: Optional[pd.DataFrame]
def __init__( 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,
price_column: str,
): ):
self.symbol_a_ = symbol_a self.symbol_a_ = symbol_a
self.symbol_b_ = symbol_b self.symbol_b_ = symbol_b
@ -102,25 +107,33 @@ class TradingPair(ABC):
) )
self.market_data_ = self.market_data_.dropna().reset_index(drop=True) 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_["tstamp"] = pd.to_datetime(self.market_data_["tstamp"])
self.market_data_ = self.market_data_.sort_values('tstamp') self.market_data_ = self.market_data_.sort_values("tstamp")
def get_begin_index(self) -> int: def get_begin_index(self) -> int:
if "trading_hours" not in self.config_: if "trading_hours" not in self.config_:
return 0 return 0
assert "timezone" in self.config_["trading_hours"] assert "timezone" in self.config_["trading_hours"]
assert "begin_session" 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() start_time = (
mask = self.market_data_['tstamp'].dt.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()) return int(self.market_data_.index[mask].min())
def get_end_index(self) -> int: def get_end_index(self) -> int:
if "trading_hours" not in self.config_: if "trading_hours" not in self.config_:
return 0 return 0
assert "timezone" in self.config_["trading_hours"] assert "timezone" in self.config_["trading_hours"]
assert "end_session" 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() end_time = (
mask = self.market_data_['tstamp'].dt.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()) return int(self.market_data_.index[mask].max())
def _transform_dataframe(self, df: pd.DataFrame) -> pd.DataFrame: def _transform_dataframe(self, df: pd.DataFrame) -> pd.DataFrame:
@ -171,7 +184,7 @@ class TradingPair(ABC):
testing_start_index = training_start_index + training_minutes testing_start_index = training_start_index + training_minutes
self.training_df_ = self.market_data_.iloc[ self.training_df_ = self.market_data_.iloc[
training_start_index:testing_start_index, : training_minutes training_start_index:testing_start_index, :training_minutes
].copy() ].copy()
assert self.training_df_ is not None assert self.training_df_ is not None
self.training_df_ = self.training_df_.dropna().reset_index(drop=True) self.training_df_ = self.training_df_.dropna().reset_index(drop=True)
@ -199,7 +212,7 @@ class TradingPair(ABC):
else: else:
# Ensure both DataFrames have the same columns and dtypes before concatenation # Ensure both DataFrames have the same columns and dtypes before concatenation
existing_trades = self.user_data_["trades"] existing_trades = self.user_data_["trades"]
# If existing trades is empty, just assign the new trades # If existing trades is empty, just assign the new trades
if len(existing_trades) == 0: if len(existing_trades) == 0:
self.user_data_["trades"] = trades.copy() self.user_data_["trades"] = trades.copy()
@ -213,22 +226,26 @@ class TradingPair(ABC):
trades[col] = pd.Timestamp.now() trades[col] = pd.Timestamp.now()
elif col in ["action", "symbol"]: elif col in ["action", "symbol"]:
trades[col] = "" trades[col] = ""
elif col in ["price", "disequilibrium", "scaled_disequilibrium"]: elif col in [
"price",
"disequilibrium",
"scaled_disequilibrium",
]:
trades[col] = 0.0 trades[col] = 0.0
elif col == "pair": elif col == "pair":
trades[col] = None trades[col] = None
else: else:
trades[col] = None trades[col] = None
# Concatenate with explicit dtypes to avoid warnings # Concatenate with explicit dtypes to avoid warnings
self.user_data_["trades"] = pd.concat( self.user_data_["trades"] = pd.concat(
[existing_trades, trades], [existing_trades, trades], ignore_index=True, copy=False
ignore_index=True,
copy=False
) )
def get_trades(self) -> pd.DataFrame: 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]: def cointegration_check(self) -> Optional[pd.DataFrame]:
print(f"***{self}*** STARTING....") print(f"***{self}*** STARTING....")
@ -237,17 +254,19 @@ class TradingPair(ABC):
curr_training_start_idx = 0 curr_training_start_idx = 0
COINTEGRATION_DATA_COLUMNS = { COINTEGRATION_DATA_COLUMNS = {
"tstamp" : "datetime64[ns]", "tstamp": "datetime64[ns]",
"pair" : "string", "pair": "string",
"eg_pvalue" : "float64", "eg_pvalue": "float64",
"johansen_lr1" : "float64", "johansen_lr1": "float64",
"johansen_cvt" : "float64", "johansen_cvt": "float64",
"eg_is_cointegrated" : "bool", "eg_is_cointegrated": "bool",
"johansen_is_cointegrated" : "bool", "johansen_is_cointegrated": "bool",
} }
# Initialize trades DataFrame with proper dtypes to avoid concatenation warnings # 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"] training_minutes = config["training_minutes"]
while True: while True:
print(curr_training_start_idx, end="\r") print(curr_training_start_idx, end="\r")
@ -271,13 +290,16 @@ class TradingPair(ABC):
def to_stop_close_conditions(self, predicted_row: pd.Series) -> bool: def to_stop_close_conditions(self, predicted_row: pd.Series) -> bool:
config = self.config_ 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 return False
if "profit" in config["stop_close_conditions"]: if "profit" in config["stop_close_conditions"]:
current_return = self._current_return(predicted_row) current_return = self._current_return(predicted_row)
# #
# print(f"time={predicted_row['tstamp']} current_return={current_return}") # print(f"time={predicted_row['tstamp']} current_return={current_return}")
# #
if current_return >= config["stop_close_conditions"]["profit"]: if current_return >= config["stop_close_conditions"]["profit"]:
print(f"STOP PROFIT: {current_return}") print(f"STOP PROFIT: {current_return}")
self.user_data_["stop_close_state"] = PairState.CLOSE_STOP_PROFIT self.user_data_["stop_close_state"] = PairState.CLOSE_STOP_PROFIT
@ -288,9 +310,10 @@ class TradingPair(ABC):
self.user_data_["stop_close_state"] = PairState.CLOSE_STOP_LOSS self.user_data_["stop_close_state"] = PairState.CLOSE_STOP_LOSS
return True return True
return False return False
def on_open_trades(self, trades: pd.DataFrame) -> None: 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 self.user_data_["open_trades"] = trades
def on_close_trades(self, trades: pd.DataFrame) -> None: def on_close_trades(self, trades: pd.DataFrame) -> None:
@ -302,20 +325,25 @@ class TradingPair(ABC):
open_trades = self.user_data_["open_trades"] open_trades = self.user_data_["open_trades"]
if len(open_trades) == 0: if len(open_trades) == 0:
return 0.0 return 0.0
def _single_instrument_return(symbol: str) -> float: def _single_instrument_return(symbol: str) -> float:
instrument_open_trades = open_trades[open_trades["symbol"] == symbol] instrument_open_trades = open_trades[open_trades["symbol"] == symbol]
instrument_open_price = 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 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.price_column_}_{symbol}"]
instrument_return = sign * (instrument_price - instrument_open_price) / instrument_open_price instrument_return = (
sign
* (instrument_price - instrument_open_price)
/ instrument_open_price
)
return float(instrument_return) * 100.0 return float(instrument_return) * 100.0
instrument_a_return = _single_instrument_return(self.symbol_a_) instrument_a_return = _single_instrument_return(self.symbol_a_)
instrument_b_return = _single_instrument_return(self.symbol_b_) 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 return 0.0
def __repr__(self) -> str: def __repr__(self) -> str:
return self.name() return self.name()
@ -328,4 +356,3 @@ class TradingPair(ABC):
# @abstractmethod # @abstractmethod
# def predicted_df(self) -> Optional[pd.DataFrame]: ... # def predicted_df(self) -> Optional[pd.DataFrame]: ...

2
lib/tools/data_loader.py
View File

@ -5,7 +5,7 @@ from typing import Dict, List, cast
import pandas as pd import pandas as pd
def load_sqlite_to_dataframe(db_path, query): def load_sqlite_to_dataframe(db_path:str, query:str) -> pd.DataFrame:
try: try:
conn = sqlite3.connect(db_path) conn = sqlite3.connect(db_path)

6218
research/notebooks/single_pair_test.ipynb
View File

File diff suppressed because one or more lines are too long

37
research/pt_backtest.py
View File

@ -3,7 +3,7 @@ import glob
import importlib import importlib
import os import os
from datetime import date, datetime from datetime import date, datetime
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional, Tuple
import pandas as pd import pandas as pd
@ -17,11 +17,13 @@ from pt_trading.results import (
from pt_trading.fit_method import PairsTradingFitMethod from pt_trading.fit_method import PairsTradingFitMethod
from pt_trading.trading_pair import TradingPair from pt_trading.trading_pair import TradingPair
DayT = str
DataFileNameT = str
def resolve_datafiles( def resolve_datafiles(
config: Dict, date_pattern: str, instruments: List[Dict[str, str]] config: Dict, date_pattern: str, instruments: List[Dict[str, str]]
) -> List[str]: ) -> List[Tuple[DayT, DataFileNameT]]:
resolved_files = [] resolved_files: List[Tuple[DayT, DataFileNameT]] = []
for inst in instruments: for inst in instruments:
pattern = date_pattern pattern = date_pattern
inst_type = inst["instrument_type"] inst_type = inst["instrument_type"]
@ -31,12 +33,17 @@ def resolve_datafiles(
if not os.path.isabs(pattern): if not os.path.isabs(pattern):
pattern = os.path.join(data_dir, f"{pattern}.mktdata.ohlcv.db") pattern = os.path.join(data_dir, f"{pattern}.mktdata.ohlcv.db")
matched_files = glob.glob(pattern) 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: else:
# Handle explicit file path # Handle explicit file path
if not os.path.isabs(pattern): if not os.path.isabs(pattern):
pattern = os.path.join(data_dir, f"{pattern}.mktdata.ohlcv.db") pattern = os.path.join(data_dir, f"{pattern}.mktdata.ohlcv.db")
resolved_files.append(pattern) resolved_files.append((date_pattern, pattern))
return sorted(list(set(resolved_files))) # Remove duplicates and sort return sorted(list(set(resolved_files))) # Remove duplicates and sort
@ -61,7 +68,7 @@ def get_instruments(args: argparse.Namespace, config: Dict) -> List[Dict[str, st
def run_backtest( def run_backtest(
config: Dict, config: Dict,
datafile: str, datafiles: List[str],
price_column: str, price_column: str,
fit_method: PairsTradingFitMethod, fit_method: PairsTradingFitMethod,
instruments: List[Dict[str, str]], instruments: List[Dict[str, str]],
@ -73,7 +80,7 @@ def run_backtest(
pairs_trades = [] pairs_trades = []
pairs = create_pairs( pairs = create_pairs(
datafile=datafile, datafiles=datafiles,
fit_method=fit_method, fit_method=fit_method,
price_column=price_column, price_column=price_column,
config=config, config=config,
@ -136,6 +143,7 @@ def main() -> None:
print("No data files found to process.") print("No data files found to process.")
return return
days = list(set([day for day, _ in datafiles]))
print(f"Found {len(datafiles)} data files to process:") print(f"Found {len(datafiles)} data files to process:")
for df in datafiles: for df in datafiles:
print(f" - {df}") print(f" - {df}")
@ -166,8 +174,9 @@ def main() -> None:
# Process each data file # Process each data file
price_column = config["price_column"] price_column = config["price_column"]
for datafile in datafiles: for day in sorted(days):
print(f"\n====== Processing {os.path.basename(datafile)} ======") md_datafiles = [datafile for md_day, datafile in datafiles if md_day == day]
print(f"\n====== Processing {day} ======")
# Process data for this file # Process data for this file
try: try:
@ -175,14 +184,14 @@ def main() -> None:
bt_results = run_backtest( bt_results = run_backtest(
config=config, config=config,
datafile=datafile, datafiles=md_datafiles,
price_column=price_column, price_column=price_column,
fit_method=fit_method, fit_method=fit_method,
instruments=instruments, instruments=instruments,
) )
# Store results with file name as key # Store results with day name as key
filename = os.path.basename(datafile) filename = os.path.basename(day)
all_results[filename] = { all_results[filename] = {
"trades": bt_results.trades.copy(), "trades": bt_results.trades.copy(),
"outstanding_positions": bt_results.outstanding_positions.copy(), "outstanding_positions": bt_results.outstanding_positions.copy(),
@ -198,12 +207,12 @@ def main() -> None:
} }
} }
) )
bt_results.store_results_in_database(args.result_db, datafile) bt_results.store_results_in_database(db_path=args.result_db, day=day)
print(f"Successfully processed {filename}") print(f"Successfully processed {filename}")
except Exception as err: except Exception as err:
print(f"Error processing {datafile}: {str(err)}") print(f"Error processing {day}: {str(err)}")
import traceback import traceback
traceback.print_exc() traceback.print_exc()

20
research/research_tools.py
View File

@ -1,6 +1,7 @@
import glob import glob
import os import os
from typing import Dict, List, Optional from typing import Dict, List, Optional
import pandas as pd
from pt_trading.fit_method import PairsTradingFitMethod from pt_trading.fit_method import PairsTradingFitMethod
@ -45,7 +46,7 @@ def resolve_datafiles(config: Dict, cli_datafiles: Optional[str] = None) -> List
def create_pairs( def create_pairs(
datafile: str, datafiles: List[str],
fit_method: PairsTradingFitMethod, fit_method: PairsTradingFitMethod,
price_column: str, price_column: str,
config: Dict, config: Dict,
@ -61,13 +62,16 @@ def create_pairs(
# Update config to use the specified instruments # Update config to use the specified instruments
config_copy = config.copy() config_copy = config.copy()
config_copy["instruments"] = instruments config_copy["instruments"] = instruments
market_data_df = load_market_data( market_data_df = pd.DataFrame()
datafile=datafile, for datafile in datafiles:
instruments=instruments, md_df = load_market_data(
db_table_name=config_copy["market_data_loading"][instruments[0]["instrument_type"]]["db_table_name"], datafile=datafile,
trading_hours=config_copy["trading_hours"], instruments=instruments,
) db_table_name=config_copy["market_data_loading"][instruments[0]["instrument_type"]]["db_table_name"],
trading_hours=config_copy["trading_hours"],
)
market_data_df = pd.concat([market_data_df, md_df])
for a_index, b_index in unique_index_pairs: for a_index, b_index in unique_index_pairs:
from research.pt_backtest import TradingPair from research.pt_backtest import TradingPair