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-14 00:41:46 +00:00
parent b24285802a
commit 3b003c7811
7 changed files with 132 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
configuration/crypto.cfg
View File

@ -7,16 +7,6 @@
"db_table_name": "md_1min_bars",
"exchange_id": "BNBSPOT",
"instrument_id_pfx": "PAIR-",
# "instruments": [
# "BTC-USDT",
# "BCH-USDT",
# "ETH-USDT",
# "LTC-USDT",
# "XRP-USDT",
# "ADA-USDT",
# "SOL-USDT",
# "DOT-USDT"
# ],
"trading_hours": {
"begin_session": "00:00:00",
"end_session": "23:59:00",
@ -29,5 +19,5 @@
"dis-equilibrium_close_trshld": 0.5,
"training_minutes": 120,
"funding_per_pair": 2000.0,
"fit_method_class": "pt_trading.fit_methods.StaticFit"
"fit_method_class": "pt_trading.fit_methods.SlidingFit"
}

119
lib/pt_trading/fit_methods.py
View File

@ -3,7 +3,6 @@ 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
@ -64,6 +63,17 @@ class StaticFit(PairsTradingFitMethod):
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"]
@ -96,11 +106,11 @@ class StaticFit(PairsTradingFitMethod):
break
open_row = predicted_df.loc[open_row_index]
open_tstamp = open_row["tstamp"]
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"]
open_px_a = open_row[f"{colname_a}"]
open_px_b = open_row[f"{colname_b}"]
abs_beta = abs(beta[1])
pred_px_b = predicted_df.loc[open_row_index][f"{colname_b}_pred"]
@ -129,9 +139,9 @@ class StaticFit(PairsTradingFitMethod):
last_row_index=last_row_index,
open_side_a=open_side_a,
open_side_b=open_side_b,
open_px_a=open_px_a,
open_px_b=open_px_b,
open_tstamp=open_tstamp,
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)
@ -205,11 +215,21 @@ class StaticFit(PairsTradingFitMethod):
),
]
# Add tuples to data frame
return pd.DataFrame(
# Add tuples to data frame with explicit dtypes to avoid concatenation warnings
df = pd.DataFrame(
trd_signal_tuples,
columns=self.TRADES_COLUMNS, # type: ignore
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
@ -232,7 +252,16 @@ class SlidingFit(PairsTradingFitMethod):
print(f"***{pair}*** STARTING....")
pair.user_data_["state"] = PairState.INITIAL
pair.user_data_["trades"] = pd.DataFrame(columns=self.TRADES_COLUMNS)
# Initialize trades DataFrame with proper dtypes to avoid concatenation warnings
pair.user_data_["trades"] = pd.DataFrame(columns=self.TRADES_COLUMNS).astype({
"time": "datetime64[ns]",
"action": "string",
"symbol": "string",
"price": "float64",
"disequilibrium": "float64",
"scaled_disequilibrium": "float64",
"pair": "object"
})
pair.user_data_["is_cointegrated"] = False
training_minutes = config["training_minutes"]
@ -255,17 +284,17 @@ class SlidingFit(PairsTradingFitMethod):
)
# outstanding positions
# last_row_index = self.curr_training_start_idx_ + training_minutes
bt_result.handle_outstanding_position(
pair=pair,
pair_result_df=pair.predicted_df_,
last_row_index=0,
open_side_a=pair.user_data_["open_side_a"],
open_side_b=pair.user_data_["open_side_b"],
open_px_a=pair.user_data_["open_px_a"],
open_px_b=pair.user_data_["open_px_b"],
open_tstamp=pair.user_data_["open_tstamp"],
)
if pair.predicted_df_ is not None:
bt_result.handle_outstanding_position(
pair=pair,
pair_result_df=pair.predicted_df_,
last_row_index=0,
open_side_a=pair.user_data_["open_side_a"],
open_side_b=pair.user_data_["open_side_b"],
open_px_a=pair.user_data_["open_px_a"],
open_px_b=pair.user_data_["open_px_b"],
open_tstamp=pair.user_data_["open_tstamp"],
)
break
try:
@ -311,7 +340,10 @@ class SlidingFit(PairsTradingFitMethod):
def _create_trading_signals(
self, pair: TradingPair, config: Dict, bt_result: BacktestResult
) -> None:
assert pair.predicted_df_ is not None
if pair.predicted_df_ is None:
print(f"{pair.market_data_.iloc[0]['tstamp']} {pair}: No predicted data")
return
open_threshold = config["dis-equilibrium_open_trshld"]
close_threshold = config["dis-equilibrium_close_trshld"]
for curr_predicted_row_idx in range(len(pair.predicted_df_)):
@ -353,6 +385,16 @@ class SlidingFit(PairsTradingFitMethod):
open_scaled_disequilibrium = open_row["scaled_disequilibrium"]
open_px_a = open_row[f"{colname_a}"]
open_px_b = open_row[f"{colname_b}"]
# Ensure scalars for handle_outstanding_position
if isinstance(open_px_a, pd.Series):
open_px_a = open_px_a.iloc[0]
if isinstance(open_px_b, pd.Series):
open_px_b = open_px_b.iloc[0]
if isinstance(open_tstamp, pd.Series):
open_tstamp = open_tstamp.iloc[0]
open_px_a = float(open_px_a)
open_px_b = float(open_px_b)
open_tstamp = pd.Timestamp(open_tstamp)
if open_scaled_disequilibrium < open_threshold:
return None
@ -402,10 +444,21 @@ class SlidingFit(PairsTradingFitMethod):
pair,
),
]
return pd.DataFrame(
# Create DataFrame with explicit dtypes to avoid concatenation warnings
df = pd.DataFrame(
trd_signal_tuples,
columns=self.TRADES_COLUMNS, # type: ignore
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 _get_close_trades(
self, pair: TradingPair, row: pd.Series, close_threshold: float
@ -449,11 +502,21 @@ class SlidingFit(PairsTradingFitMethod):
),
]
# Add tuples to data frame
return pd.DataFrame(
# Add tuples to data frame with explicit dtypes to avoid concatenation warnings
df = pd.DataFrame(
trd_signal_tuples,
columns=self.TRADES_COLUMNS, # type: ignore
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:
self.curr_training_start_idx_ = 0

6
lib/pt_trading/results.py
View File

@ -41,6 +41,12 @@ def create_result_database(db_path: str) -> None:
Create the SQLite database and required tables if they don't exist.
"""
try:
# Create directory if it doesn't exist
db_dir = os.path.dirname(db_path)
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()

35
lib/pt_trading/trading_pair.py
View File

@ -179,10 +179,39 @@ class TradingPair:
return result
def add_trades(self, trades: pd.DataFrame) -> None:
if self.user_data_["trades"] is None:
self.user_data_["trades"] = pd.DataFrame(trades)
if self.user_data_["trades"] is None or len(self.user_data_["trades"]) == 0:
# If trades is empty or None, just assign the new trades directly
self.user_data_["trades"] = trades.copy()
else:
self.user_data_["trades"] = pd.concat([self.user_data_["trades"], pd.DataFrame(trades)], ignore_index=True)
# 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()
else:
# Ensure both DataFrames have the same columns
if set(existing_trades.columns) != set(trades.columns):
# Add missing columns to trades with appropriate default values
for col in existing_trades.columns:
if col not in trades.columns:
if col == "time":
trades[col] = pd.Timestamp.now()
elif col in ["action", "symbol"]:
trades[col] = ""
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
)
def get_trades(self) -> pd.DataFrame:
return self.user_data_["trades"] if "trades" in self.user_data_ else pd.DataFrame()

11
research/notebooks/pt_pair_backtest.ipynb
View File

@ -936,7 +936,7 @@
},
{
"cell_type": "code",
"execution_count": 10,
"execution_count": null,
"metadata": {},
"outputs": [
{
@ -1075,15 +1075,6 @@
" linestyle=':', alpha=0.7)\n",
" axes[1].axhline(y=0, color='black', linestyle='-', alpha=0.5, linewidth=0.5)\n",
"\n",
" # if pair_trades is not None and len(pair_trades) > 0:\n",
" # # Show trading signals over time\n",
" # trade_times = pair_trades['time'].values\n",
" # trade_actions = pair_trades['action'].values\n",
" \n",
" # for i, (time, action) in enumerate(zip(trade_times, trade_actions)):\n",
" # color = 'red' if 'BUY' in action else 'blue'\n",
" # axes[1].scatter(time, i, color=color, alpha=0.8, s=50)\n",
" \n",
"\n",
" axes[1].set_title('Testing Period: Scaled Dis-equilibrium with Trading Thresholds')\n",
" axes[1].set_ylabel('Scaled Dis-equilibrium')\n",

BIN
researchresults/equity/20250714_003409.equity_results.db Normal file
View File

Binary file not shown.

1
sync_visualization.py Normal file
View File

@ -0,0 +1 @@