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 2026-01-12 21:26:15 +00:00
parent bd6cf1d4d0
commit 3b5afcbaa0
15 changed files with 1205 additions and 568 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
.vscode/launch.json vendored
View File

@ -21,18 +21,23 @@
"name": "-------- Live Pair Trading --------", "name": "-------- Live Pair Trading --------",
}, },
{ {
"name": "PAIRS TRADER", "name": "PAIR TRADER",
"type": "debugpy", "type": "debugpy",
"request": "launch", "request": "launch",
"python": "/home/oleg/.pyenv/python3.12-venv/bin/python", "python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
"program": "${workspaceFolder}/bin/pairs_trader.py", "program": "${workspaceFolder}/apps/pair_trader.py",
"console": "integratedTerminal", "console": "integratedTerminal",
"env": { "env": {
"PYTHONPATH": "${workspaceFolder}/.." "PYTHONPATH": "${workspaceFolder}/..",
"CONFIG_SERVICE": "cloud16.cvtt.vpn:6789",
"MODEL_CONFIG": "vecm"
}, },
"args": [ "args": [
"--config=${workspaceFolder}/configuration/pairs_trader.cfg", // "--config=${workspaceFolder}/configuration/pair_trader.cfg",
"--pair=PAIR-ADA-USDT:BNBSPOT,PAIR-SOL-USDT:BNBSPOT", "--config=http://cloud16.cvtt.vpn:6789/apps/pairs_trading/pair_trader",
"--book_id=TEST_BOOK_20250818",
"--instrument_A=COINBASE_AT:PAIR-ADA-USD",
"--instrument_B=COINBASE_AT:PAIR-SOL-USD",
], ],
}, },
{ {
@ -45,14 +50,15 @@
"python": "/home/oleg/.pyenv/python3.12-venv/bin/python", "python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
"program": "${workspaceFolder}/research/backtest.py", "program": "${workspaceFolder}/research/backtest.py",
"args": [ "args": [
"--config=${workspaceFolder}/configuration/vecm-opt.cfg", "--config=http://cloud16.cvtt.vpn:6789/apps/pairs_trading/backtest",
"--instruments=CRYPTO:BNBSPOT:PAIR-ADA-USDT,CRYPTO:BNBSPOT:PAIR-SOL-USDT", "--instruments=CRYPTO:BNBSPOT:PAIR-ADA-USDT,CRYPTO:BNBSPOT:PAIR-SOL-USDT",
"--date_pattern=20250910", "--date_pattern=20250911",
"--result_db=${workspaceFolder}/research/results/crypto/%T.vecm-opt.ADA-SOL.20250605.crypto_results.db", "--result_db=${workspaceFolder}/research/results/crypto/%T.vecm-opt.ADA-SOL.20250605.crypto_results.db",
], ],
"env": { "env": {
"PYTHONPATH": "${workspaceFolder}/..", "PYTHONPATH": "${workspaceFolder}/..",
"CONFIG_SERVICE": "cloud16.cvtt.vpn:6789" "CONFIG_SERVICE": "cloud16.cvtt.vpn:6789",
"MODEL_CONFIG": "vecm-opt"
}, },
"console": "integratedTerminal" "console": "integratedTerminal"
}, },

74
apps/pairs_trader.py → apps/pair_trader.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations from __future__ import annotations
from typing import Callable, Coroutine, List import asyncio
from typing import Callable, Coroutine, Dict, List
import aiohttp.web as web import aiohttp.web as web
from cvttpy_tools.app import App from cvttpy_tools.app import App
@ -16,9 +17,7 @@ from cvttpy_trading.trading.instrument import ExchangeInstrument
from cvttpy_trading.trading.mkt_data.md_summary import MdTradesAggregate from cvttpy_trading.trading.mkt_data.md_summary import MdTradesAggregate
from cvttpy_trading.trading.exchange_config import ExchangeAccounts from cvttpy_trading.trading.exchange_config import ExchangeAccounts
# --- # ---
from pairs_trading.lib.pt_strategy.live.live_strategy import PtLiveStrategy
from pairs_trading.lib.live.mkt_data_client import CvttRestMktDataClient from pairs_trading.lib.live.mkt_data_client import CvttRestMktDataClient
from pairs_trading.lib.live.ti_sender import TradingInstructionsSender
''' '''
config http://cloud16.cvtt.vpn/apps/pairs_trading config http://cloud16.cvtt.vpn/apps/pairs_trading
@ -27,27 +26,36 @@ config http://cloud16.cvtt.vpn/apps/pairs_trading
HistMdCbT = Callable[[List[MdTradesAggregate]], Coroutine] HistMdCbT = Callable[[List[MdTradesAggregate]], Coroutine]
UpdateMdCbT = Callable[[MdTradesAggregate], Coroutine] UpdateMdCbT = Callable[[MdTradesAggregate], Coroutine]
class PairsTrader(NamedObject): class PairTrader(NamedObject):
config_: CvttAppConfig config_: CvttAppConfig
instruments_: List[ExchangeInstrument] instruments_: List[ExchangeInstrument]
book_id_: BookIdT book_id_: BookIdT
live_strategy_: PtLiveStrategy live_strategy_: "PtLiveStrategy" #type: ignore
ti_sender_: "TradingInstructionsSender" #type: ignore
pricer_client_: CvttRestMktDataClient pricer_client_: CvttRestMktDataClient
ti_sender_: TradingInstructionsSender
rest_service_: RestService rest_service_: RestService
latest_history_: Dict[ExchangeInstrument, List[MdTradesAggregate]]
def __init__(self) -> None: def __init__(self) -> None:
self.instruments_ = [] self.instruments_ = []
self.latest_history_ = {}
App.instance().add_cmdline_arg( App.instance().add_cmdline_arg(
"--pair", "--instrument_A",
type=str, type=str,
required=True, required=True,
help=( help=(
"Comma-separated pair of instrument symbols" " Instrument A in pair (e.g., COINBASE_AT:PAIR-BTC-USD)"
" with exchange config name" ),
" (e.g., PAIR-BTC-USD:BNBSPOT,PAIR-ETH-USD:BNBSPOT)" )
App.instance().add_cmdline_arg(
"--instrument_B",
type=str,
required=True,
help=(
" Instrument B in pair (e.g., COINBASE_AT:PAIR-ETH-USD)"
), ),
) )
@ -65,21 +73,21 @@ class PairsTrader(NamedObject):
self.book_id_ = App.instance().get_argument(name="book_id") self.book_id_ = App.instance().get_argument(name="book_id")
# ------- PARSE INSTRUMENTS ------- # ------- PARSE INSTRUMENTS -------
instr_str = App.instance().get_argument("pair", "") instr_list: List[str] = []
if not instr_str: instr_str = App.instance().get_argument("instrument_A", "")
raise ValueError("Pair is required") assert instr_str != "", "Missing insrument A"
instr_list = instr_str.split(",") instr_list.append(instr_str)
instr_str = App.instance().get_argument("instrument_B", "")
assert len(instr_list) == 2, "Only two instruments are supported" assert instr_str != "", "Missing insrument B"
instr_list.append(instr_str)
for instr in instr_list: for instr in instr_list:
instr_parts = instr.split(":") instr_parts = instr.split(":")
if len(instr_parts) != 2: if len(instr_parts) != 2:
raise ValueError(f"Invalid pair format: {instr}") raise ValueError(f"Invalid pair format: {instr}")
instrument_id = instr_parts[0] exch_acct = instr_parts[0]
exch_acct = instr_parts[1] instrument_id = instr_parts[1]
exch_inst = ExchangeAccounts.instance().get_exchange_instrument(exch_acct=exch_acct, instrument_id=instrument_id) exch_inst = ExchangeAccounts.instance().get_exchange_instrument(exch_acct=exch_acct, instrument_id=instrument_id)
assert exch_inst is not None, f"No ExchangeInstrument for {instr}" assert exch_inst is not None, f"No ExchangeInstrument for {instr}"
exch_inst.user_data_["exch_acct"] = exch_acct exch_inst.user_data_["exch_acct"] = exch_acct
self.instruments_.append(exch_inst) self.instruments_.append(exch_inst)
@ -87,20 +95,24 @@ class PairsTrader(NamedObject):
Log.info(f"{self.fname()} Instruments: {self.instruments_[0].details_short()} <==> {self.instruments_[1].details_short()}") Log.info(f"{self.fname()} Instruments: {self.instruments_[0].details_short()} <==> {self.instruments_[1].details_short()}")
# ------- CREATE STRATEGY ------- # ------- CREATE STRATEGY -------
strategy_config = self.config_.get_subconfig("strategy_config", Config({})) from pairs_trading.lib.pt_strategy.live.live_strategy import PtLiveStrategy
strategy_config = CvttAppConfig.instance() #self.config_.get_subconfig("strategy_config", Config({}))
self.live_strategy_ = PtLiveStrategy( self.live_strategy_ = PtLiveStrategy(
config=strategy_config, config=strategy_config,
pairs_trader=self, pairs_trader=self,
) )
Log.info(f"{self.fname()} Strategy created: {self.live_strategy_}") Log.info(f"{self.fname()} Strategy created: {self.live_strategy_}")
model_name = self.config_.get_value("model/name", "?model/name?")
self.config_.set_value("strategy_id", f"{self.live_strategy_.__class__.__name__}:{model_name}")
# # ------- CREATE PRICER CLIENT ------- # # ------- CREATE PRICER CLIENT -------
self.pricer_client_ = CvttRestMktDataClient(config=self.config_) self.pricer_client_ = CvttRestMktDataClient(config=self.config_)
Log.info(f"{self.fname()} MD client created: {self.pricer_client_}") Log.info(f"{self.fname()} MD client created: {self.pricer_client_}")
# ------- CREATE TRADER CLIENT ------- # ------- CREATE TRADER CLIENT -------
from pairs_trading.lib.live.ti_sender import TradingInstructionsSender
self.ti_sender_ = TradingInstructionsSender(config=self.config_, pairs_trader=self) self.ti_sender_ = TradingInstructionsSender(config=self.config_, pairs_trader=self)
Log.info(f"{self.fname()} TI sebder created: {self.ti_sender_}") Log.info(f"{self.fname()} TI sender created: {self.ti_sender_}")
# # ------- CREATE REST SERVER ------- # # ------- CREATE REST SERVER -------
self.rest_service_ = RestService( self.rest_service_ = RestService(
@ -115,6 +127,7 @@ class PairsTrader(NamedObject):
) )
async def subscribe_md(self) -> None: async def subscribe_md(self) -> None:
from functools import partial
for exch_inst in self.instruments_: for exch_inst in self.instruments_:
exch_acct = exch_inst.user_data_.get("exch_acct", "?exch_acct?") exch_acct = exch_inst.user_data_.get("exch_acct", "?exch_acct?")
instrument_id = exch_inst.instrument_id() instrument_id = exch_inst.instrument_id()
@ -124,12 +137,19 @@ class PairsTrader(NamedObject):
instrument_id=instrument_id, instrument_id=instrument_id,
interval_sec=self.live_strategy_.interval_sec(), interval_sec=self.live_strategy_.interval_sec(),
history_depth_sec=self.live_strategy_.history_depth_sec(), history_depth_sec=self.live_strategy_.history_depth_sec(),
callback=self._on_md_summary callback=partial(self._on_md_summary, exch_inst=exch_inst)
) )
async def _on_md_summary(self, history: List[MdTradesAggregate]) -> None: async def _on_md_summary(self, history: List[MdTradesAggregate], exch_inst: ExchangeInstrument) -> None:
# Snapshot or update? # URGENT before calling stragegy, make sure that **BOTH** instruments market data is combined.
await self.live_strategy_.on_mkt_data_hist_snapshot(hist_aggr=history) Log.info(f"DEBUG got {exch_inst.details_short()} data")
self.latest_history_[exch_inst] = history
if len(self.latest_history_) == 2:
from itertools import chain
all_aggrs = sorted(list(chain.from_iterable(self.latest_history_.values())), key=lambda X: X.time_ns_)
await self.live_strategy_.on_mkt_data_hist_snapshot(hist_aggr=all_aggrs)
self.latest_history_ = {}
async def _on_api_request(self, request: web.Request) -> web.Response: async def _on_api_request(self, request: web.Request) -> web.Response:
# TODO choose pair # TODO choose pair
@ -139,10 +159,12 @@ class PairsTrader(NamedObject):
async def run(self) -> None: async def run(self) -> None:
Log.info(f"{self.fname()} ...") Log.info(f"{self.fname()} ...")
while True:
await asyncio.sleep(0.1)
pass pass
if __name__ == "__main__": if __name__ == "__main__":
App() App()
CvttAppConfig() CvttAppConfig()
PairsTrader() PairTrader()
App.instance().run() App.instance().run()

46
configuration/backtest.cfg Normal file
View File

@ -0,0 +1,46 @@
{
"refdata": {
"assets": @inc=http://@env{CONFIG_SERVICE}/refdata/assets
, "instruments": @inc=http://@env{CONFIG_SERVICE}/refdata/instruments
, "exchange_instruments": @inc=http://@env{CONFIG_SERVICE}/refdata/exchange_instruments
, "dynamic_instrument_exchanges": ["ALPACA"]
, "exchanges": @inc=http://@env{CONFIG_SERVICE}/refdata/exchanges
},
"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,
# ====== Model =======
"model": @inc=http://@env{CONFIG_SERVICE}/apps/common/models/@env{MODEL_CONFIG}
# ====== Trading =======
"execution_price": {
"column": "vwap",
"shift": 1,
},
# ====== 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": "7:30:00",
"end_session": "18:30:00",
}
}

2
configuration/__DELETE__/pairs_trader.cfg → configuration/pair_trader.cfg
View File

@ -1,5 +1,5 @@
{ {
"strategy_config": @inc=file:///home/oleg/develop/pairs_trading/configuration/ols.cfg "strategy_config": @inc=file:///home/oleg/develop/pairs_trading/configuration/vecm-opt.cfg
"pricer_config": { "pricer_config": {
"pricer_url": "ws://localhost:12346/ws", "pricer_url": "ws://localhost:12346/ws",
"history_depth_sec": 86400 #"60*60*24", # use simpleeval "history_depth_sec": 86400 #"60*60*24", # use simpleeval

68
configuration/vecm-opt.cfg
View File

@ -1,26 +1,26 @@
{ {
"refdata": { # "refdata": {
"assets": @inc=http://@env{CONFIG_SERVICE}/refdata/assets # "assets": @inc=http://@env{CONFIG_SERVICE}/refdata/assets
, "instruments": @inc=http://@env{CONFIG_SERVICE}/refdata/instruments # , "instruments": @inc=http://@env{CONFIG_SERVICE}/refdata/instruments
, "exchange_instruments": @inc=http://@env{CONFIG_SERVICE}/refdata/exchange_instruments # , "exchange_instruments": @inc=http://@env{CONFIG_SERVICE}/refdata/exchange_instruments
, "dynamic_instrument_exchanges": ["ALPACA"] # , "dynamic_instrument_exchanges": ["ALPACA"]
, "exchanges": @inc=http://@env{CONFIG_SERVICE}/refdata/exchanges # , "exchanges": @inc=http://@env{CONFIG_SERVICE}/refdata/exchanges
}, # },
"market_data_loading": { # "market_data_loading": {
"CRYPTO": { # "CRYPTO": {
"data_directory": "./data/crypto", # "data_directory": "./data/crypto",
"db_table_name": "md_1min_bars", # "db_table_name": "md_1min_bars",
"instrument_id_pfx": "PAIR-", # "instrument_id_pfx": "PAIR-",
}, # },
"EQUITY": { # "EQUITY": {
"data_directory": "./data/equity", # "data_directory": "./data/equity",
"db_table_name": "md_1min_bars", # "db_table_name": "md_1min_bars",
"instrument_id_pfx": "STOCK-", # "instrument_id_pfx": "STOCK-",
} # }
}, # },
# ====== Funding ====== # # ====== Funding ======
"funding_per_pair": 2000.0, # "funding_per_pair": 2000.0,
# ====== Trading Parameters ====== # ====== Trading Parameters ======
"stat_model_price": "close", # "vwap" "stat_model_price": "close", # "vwap"
@ -39,18 +39,18 @@
"min_training_size": 60, "min_training_size": 60,
"max_training_size": 150, "max_training_size": 150,
# ====== Stop Conditions ====== # # ====== Stop Conditions ======
"stop_close_conditions": { # "stop_close_conditions": {
"profit": 2.0, # "profit": 2.0,
"loss": -0.5 # "loss": -0.5
} # }
# ====== End of Session Closeout ====== # # ====== End of Session Closeout ======
"close_outstanding_positions": true, # "close_outstanding_positions": true,
# "close_outstanding_positions": false, # # "close_outstanding_positions": false,
"trading_hours": { # "trading_hours": {
"timezone": "America/New_York", # "timezone": "America/New_York",
"begin_session": "7:30:00", # "begin_session": "7:30:00",
"end_session": "18:30:00", # "end_session": "18:30:00",
} # }
} }

5
lib/live/mkt_data_client.py
View File

@ -151,6 +151,7 @@ class MdSummaryCollector(NamedObject):
return MdSummary.from_REST_response(response=response) return MdSummary.from_REST_response(response=response)
def get_last(self) -> Optional[MdSummary]: def get_last(self) -> Optional[MdSummary]:
Log.info(f"{self.fname()}: for {self.exch_inst_.details_short()}")
rqst_data = self.rqst_data() rqst_data = self.rqst_data()
rqst_data["history_depth_sec"] = self.interval_sec_ * 2 rqst_data["history_depth_sec"] = self.interval_sec_ * 2
response: requests.Response = self.sender_.send_post( response: requests.Response = self.sender_.send_post(
@ -186,10 +187,12 @@ class MdSummaryCollector(NamedObject):
def set_timer(self): def set_timer(self):
if self.timer_: if self.timer_:
self.timer_.cancel() self.timer_.cancel()
start_in = self.next_load_time() - current_seconds()
self.timer_ = Timer( self.timer_ = Timer(
start_in_sec=(self.next_load_time() - current_seconds()), start_in_sec=start_in,
func=self._load_new, func=self._load_new,
) )
Log.info(f"{self.fname()} Timer for {self.exch_inst_.details_short()} is set to run in {start_in} sec")
def next_load_time(self) -> NanosT: def next_load_time(self) -> NanosT:
curr_sec = int(current_seconds()) curr_sec = int(current_seconds())

14
lib/live/ti_sender.py
View File

@ -10,13 +10,13 @@ from cvttpy_tools.logger import Log
from cvttpy_trading.trading.trading_instructions import TradingInstructions from cvttpy_trading.trading.trading_instructions import TradingInstructions
# --- # ---
from pairs_trading.lib.live.rest_client import RESTSender from pairs_trading.lib.live.rest_client import RESTSender
from pairs_trading.apps.pairs_trader import PairsTrader from pairs_trading.apps.pair_trader import PairTrader
class TradingInstructionsSender(NamedObject): class TradingInstructionsSender(NamedObject):
config_: Config config_: Config
sender_: RESTSender sender_: RESTSender
pairs_trader_: PairsTrader pairs_trader_: PairTrader
class TradingInstType(str, Enum): class TradingInstType(str, Enum):
TARGET_POSITION = "TARGET_POSITION" TARGET_POSITION = "TARGET_POSITION"
@ -24,13 +24,7 @@ class TradingInstructionsSender(NamedObject):
MARKET_MAKING = "MARKET_MAKING" MARKET_MAKING = "MARKET_MAKING"
NONE = "NONE" NONE = "NONE"
# config_: Config def __init__(self, config: Config, pairs_trader: PairTrader) -> None:
# ti_method_: str
# ti_url_: str
# health_check_method_: str
# health_check_url_: str
def __init__(self, config: Config, pairs_trader: PairsTrader) -> None:
self.config_ = config self.config_ = config
base_url = self.config_.get_value("cvtt_base_url", default="") base_url = self.config_.get_value("cvtt_base_url", default="")
assert base_url assert base_url
@ -45,7 +39,7 @@ class TradingInstructionsSender(NamedObject):
async def send_trading_instructions(self, ti: TradingInstructions) -> None: async def send_trading_instructions(self, ti: TradingInstructions) -> None:
Log.info(f"{self.fname()}: sending {ti=}")
response: requests.Response = self.sender_.send_post( response: requests.Response = self.sender_.send_post(
endpoint="trading_instructions", post_body=ti.to_dict() endpoint="trading_instructions", post_body=ti.to_dict()
) )

49
lib/pt_strategy/live/live_strategy.py
View File

@ -22,7 +22,7 @@ from cvttpy_trading.trading.trading_instructions import TargetPositionSignal
from pairs_trading.lib.pt_strategy.model_data_policy import ModelDataPolicy from pairs_trading.lib.pt_strategy.model_data_policy import ModelDataPolicy
from pairs_trading.lib.pt_strategy.pt_model import Prediction from pairs_trading.lib.pt_strategy.pt_model import Prediction
from pairs_trading.lib.pt_strategy.trading_pair import LiveTradingPair from pairs_trading.lib.pt_strategy.trading_pair import LiveTradingPair
from pairs_trading.apps.pairs_trader import PairsTrader from pairs_trading.apps.pair_trader import PairTrader
from pairs_trading.lib.pt_strategy.pt_market_data import LiveMarketData from pairs_trading.lib.pt_strategy.pt_market_data import LiveMarketData
@ -37,7 +37,7 @@ class PtLiveStrategy(NamedObject):
trading_pair_: LiveTradingPair trading_pair_: LiveTradingPair
model_data_policy_: ModelDataPolicy model_data_policy_: ModelDataPolicy
pairs_trader_: PairsTrader pairs_trader_: PairTrader
# for presentation: history of prediction values and trading signals # for presentation: history of prediction values and trading signals
predictions_df_: pd.DataFrame predictions_df_: pd.DataFrame
@ -46,22 +46,28 @@ class PtLiveStrategy(NamedObject):
def __init__( def __init__(
self, self,
config: Config, config: Config,
pairs_trader: PairsTrader, pairs_trader: PairTrader,
): ):
# import copy
# self.config_ = Config(json_src=copy.deepcopy(config.data()))
self.config_ = config
self.pairs_trader_ = pairs_trader self.pairs_trader_ = pairs_trader
self.trading_pair_ = LiveTradingPair( self.trading_pair_ = LiveTradingPair(
config=config, config=config,
instruments=self.pairs_trader_.instruments_, instruments=self.pairs_trader_.instruments_,
) )
self.model_data_policy_ = ModelDataPolicy.create(
self.config_,
is_real_time=True,
pair=self.trading_pair_,
)
assert (
self.model_data_policy_ is not None
), f"{self.fname()}: Unable to create ModelDataPolicy"
self.predictions_df_ = pd.DataFrame() self.predictions_df_ = pd.DataFrame()
self.trading_signals_df_ = pd.DataFrame() self.trading_signals_df_ = pd.DataFrame()
# self.book_ = book
import copy
# modified config must be passed to PtMarketData
self.config_ = Config(json_src=copy.deepcopy(config.data()))
self.instruments_ = self.pairs_trader_.instruments_ self.instruments_ = self.pairs_trader_.instruments_
@ -71,25 +77,27 @@ class PtLiveStrategy(NamedObject):
async def _on_config(self) -> None: async def _on_config(self) -> None:
self.interval_sec_ = self.config_.get_value("interval_sec", 0) self.interval_sec_ = self.config_.get_value("interval_sec", 0)
assert self.interval_sec_ > 0, "interval_sec cannot be 0"
self.history_depth_sec_ = ( self.history_depth_sec_ = (
self.config_.get_value("history_depth_hours", 0) * SecPerHour self.config_.get_value("history_depth_hours", 0) * SecPerHour
) )
assert self.history_depth_sec_ > 0, "history_depth_hours cannot be 0"
await self.pairs_trader_.subscribe_md() await self.pairs_trader_.subscribe_md()
self.open_threshold_ = self.config_.get_value( self.open_threshold_ = self.config_.get_value(
"dis-equilibrium_open_trshld", 0.0 "model/disequilibrium/open_trshld", 0.0
) )
self.close_threshold_ = self.config_.get_value( self.close_threshold_ = self.config_.get_value(
"dis-equilibrium_close_trshld", 0.0 "model/disequilibrium/close_trshld", 0.0
) )
assert ( assert (
self.open_threshold_ > 0 self.open_threshold_ > 0
), "dis-equilibrium_open_trshld must be greater than 0" ), "disequilibrium/open_trshld must be greater than 0"
assert ( assert (
self.close_threshold_ > 0 self.close_threshold_ > 0
), "dis-equilibrium_close_trshld must be greater than 0" ), "disequilibrium/close_trshld must be greater than 0"
def __repr__(self) -> str: def __repr__(self) -> str:
return f"{self.classname()}: trading_pair={self.trading_pair_}, mdp={self.model_data_policy_.__class__.__name__}, " return f"{self.classname()}: trading_pair={self.trading_pair_}, mdp={self.model_data_policy_.__class__.__name__}, "
@ -106,16 +114,8 @@ class PtLiveStrategy(NamedObject):
return return
self.trading_pair_.market_data_ = market_data_df self.trading_pair_.market_data_ = market_data_df
self.model_data_policy_ = ModelDataPolicy.create(
self.config_,
is_real_time=True,
pair=self.trading_pair_,
mkt_data=market_data_df,
)
assert (
self.model_data_policy_ is not None
), f"{self.fname()}: Unable to create ModelDataPolicy"
Log.info(f"{self.fname()}: Running prediction for pair: {self.trading_pair_}")
prediction = self.trading_pair_.run( prediction = self.trading_pair_.run(
market_data_df, self.model_data_policy_.advance() market_data_df, self.model_data_policy_.advance()
) )
@ -132,13 +132,16 @@ class PtLiveStrategy(NamedObject):
await self._send_trading_instructions(trading_instructions) await self._send_trading_instructions(trading_instructions)
def _is_md_actual(self, hist_aggr: List[MdTradesAggregate]) -> bool: def _is_md_actual(self, hist_aggr: List[MdTradesAggregate]) -> bool:
curr_ns = current_nanoseconds()
LAG_THRESHOLD = 5 * NanoPerSec LAG_THRESHOLD = 5 * NanoPerSec
if len(hist_aggr) == 0: if len(hist_aggr) == 0:
Log.warning(f"{self.fname()} list of aggregates IS EMPTY") Log.warning(f"{self.fname()} list of aggregates IS EMPTY")
return False return False
# MAYBE check market data length # MAYBE check market data length
if current_nanoseconds() - hist_aggr[-1].time_ns_ > LAG_THRESHOLD: lag_ns = curr_ns - hist_aggr[-1].time_ns_
if lag_ns > LAG_THRESHOLD:
Log.warning(f"{self.fname()} {hist_aggr[-1].exch_inst_.details_short()} Lagging {int(lag_ns/NanoPerSec)} seconds")
return False return False
return True return True

16
lib/pt_strategy/model_data_policy.py
View File

@ -25,7 +25,7 @@ class ModelDataPolicy(ABC):
def __init__(self, config: Config, *args: Any, **kwargs: Any): def __init__(self, config: Config, *args: Any, **kwargs: Any):
self.config_ = config self.config_ = config
self.current_data_params_ = DataWindowParams( self.current_data_params_ = DataWindowParams(
training_size_=config.get_value("training_size", 120), training_size_=config.get_value("model/training_size", 120),
training_start_index_=0, training_start_index_=0,
) )
self.count_ = 0 self.count_ = 0
@ -34,6 +34,7 @@ class ModelDataPolicy(ABC):
@abstractmethod @abstractmethod
def advance(self, mkt_data_df: Optional[pd.DataFrame] = None) -> DataWindowParams: def advance(self, mkt_data_df: Optional[pd.DataFrame] = None) -> DataWindowParams:
self.count_ += 1 self.count_ += 1
if not self.is_real_time_:
print(self.count_, end="\r") print(self.count_, end="\r")
return self.current_data_params_ return self.current_data_params_
@ -41,7 +42,7 @@ class ModelDataPolicy(ABC):
def create(config: Config, *args: Any, **kwargs: Any) -> ModelDataPolicy: def create(config: Config, *args: Any, **kwargs: Any) -> ModelDataPolicy:
import importlib import importlib
model_data_policy_class_name = config.get_value("model_data_policy_class", None) model_data_policy_class_name = config.get_value("model/model_data_policy_class", None)
assert model_data_policy_class_name is not None assert model_data_policy_class_name is not None
module_name, class_name = model_data_policy_class_name.rsplit(".", 1) module_name, class_name = model_data_policy_class_name.rsplit(".", 1)
module = importlib.import_module(module_name) module = importlib.import_module(module_name)
@ -59,7 +60,9 @@ class RollingWindowDataPolicy(ModelDataPolicy):
def advance(self, mkt_data_df: Optional[pd.DataFrame] = None) -> DataWindowParams: def advance(self, mkt_data_df: Optional[pd.DataFrame] = None) -> DataWindowParams:
super().advance(mkt_data_df) super().advance(mkt_data_df)
if self.is_real_time_: if self.is_real_time_:
self.current_data_params_.training_start_index_ = -self.current_data_params_.training_size_ self.current_data_params_.training_start_index_ = 0
if mkt_data_df and len(mkt_data_df) > self.curren_data_params_.training_size_:
self.current_data_params_.training_start_index_ = -self.curren_data_params_.training_size_
else: else:
self.current_data_params_.training_start_index_ += 1 self.current_data_params_.training_start_index_ += 1
return self.current_data_params_ return self.current_data_params_
@ -79,11 +82,10 @@ class OptimizedWndDataPolicy(ModelDataPolicy, ABC):
assert ( assert (
kwargs.get("pair") is not None kwargs.get("pair") is not None
), "pair must be provided" ), "pair must be provided"
assert ( assert (config.key_exists("model/max_training_size") and config.key_exists("model/min_training_size")
"min_training_size" in config.data() and "max_training_size" in config.data()
), "min_training_size and max_training_size must be provided" ), "min_training_size and max_training_size must be provided"
self.min_training_size_ = cast(int, config.get_value("min_training_size")) self.min_training_size_ = cast(int, config.get_value("model/min_training_size"))
self.max_training_size_ = cast(int, config.get_value("max_training_size")) self.max_training_size_ = cast(int, config.get_value("model/max_training_size"))
from pairs_trading.lib.pt_strategy.trading_pair import TradingPair from pairs_trading.lib.pt_strategy.trading_pair import TradingPair
self.pair_ = cast(TradingPair, kwargs.get("pair")) self.pair_ = cast(TradingPair, kwargs.get("pair"))

2
lib/pt_strategy/pt_market_data.py
View File

@ -30,7 +30,7 @@ class PtMarketData(NamedObject, ABC):
self.config_ = config self.config_ = config
self.origin_mkt_data_df_ = pd.DataFrame() self.origin_mkt_data_df_ = pd.DataFrame()
self.market_data_df_ = pd.DataFrame() self.market_data_df_ = pd.DataFrame()
self.stat_model_price_ = self.config_.get_value("stat_model_price") self.stat_model_price_ = self.config_.get_value("model/stat_model_price")
self.instruments_ = instruments self.instruments_ = instruments
assert len(self.instruments_) > 0, "No instruments found in config" assert len(self.instruments_) > 0, "No instruments found in config"

2
lib/pt_strategy/pt_model.py
View File

@ -19,7 +19,7 @@ class PairsTradingModel(ABC):
def create(config: Config) -> PairsTradingModel: def create(config: Config) -> PairsTradingModel:
import importlib import importlib
model_class_name = config.get_value("model_class", None) model_class_name = config.get_value("model/model_class", None)
assert model_class_name is not None assert model_class_name is not None
module_name, class_name = model_class_name.rsplit(".", 1) module_name, class_name = model_class_name.rsplit(".", 1)
module = importlib.import_module(module_name) module = importlib.import_module(module_name)

4
lib/pt_strategy/research_strategy.py
View File

@ -95,8 +95,8 @@ class PtResearchStrategy:
pair = self.trading_pair_ pair = self.trading_pair_
trades = None trades = None
open_threshold = self.config_.get_value("dis-equilibrium_open_trshld") open_threshold = self.config_.get_value("model/disequilibrium/open_trshld")
close_threshold = self.config_.get_value("dis-equilibrium_close_trshld") close_threshold = self.config_.get_value("model/disequilibrium/close_trshld")
scaled_disequilibrium = prediction.scaled_disequilibrium_ scaled_disequilibrium = prediction.scaled_disequilibrium_
abs_scaled_disequilibrium = abs(scaled_disequilibrium) abs_scaled_disequilibrium = abs(scaled_disequilibrium)

5
lib/pt_strategy/trading_pair.py
View File

@ -50,11 +50,11 @@ class TradingPair(NamedObject, ABC):
self.instruments_ = instruments self.instruments_ = instruments
self.instruments_[0].user_data_["symbol"] = instruments[0].instrument_id().split("-", 1)[1] self.instruments_[0].user_data_["symbol"] = instruments[0].instrument_id().split("-", 1)[1]
self.instruments_[1].user_data_["symbol"] = instruments[1].instrument_id().split("-", 1)[1] self.instruments_[1].user_data_["symbol"] = instruments[1].instrument_id().split("-", 1)[1]
self.stat_model_price_ = config.get_value("model/stat_model_price")
def run(self, market_data: pd.DataFrame, data_params: DataWindowParams) -> Prediction: # type: ignore[assignment] def run(self, market_data: pd.DataFrame, data_params: DataWindowParams) -> Prediction: # type: ignore[assignment]
self.market_data_ = market_data[ self.market_data_ = market_data[
data_params.training_start_index_ : data_params.training_start_index_ data_params.training_start_index_ : data_params.training_start_index_ + data_params.training_size_
+ data_params.training_size_
] ]
return self.model_.predict(pair=self) return self.model_.predict(pair=self)
@ -93,7 +93,6 @@ class ResearchTradingPair(TradingPair):
assert len(instruments) == 2, "Trading pair must have exactly 2 instruments" assert len(instruments) == 2, "Trading pair must have exactly 2 instruments"
super().__init__(config=config, instruments=instruments) super().__init__(config=config, instruments=instruments)
self.stat_model_price_ = config.get_value("stat_model_price")
self.user_data_ = { self.user_data_ = {
"state": PairState.INITIAL, "state": PairState.INITIAL,
} }

16
lib/tools/viz/viz_trades.py
View File

@ -105,8 +105,8 @@ def visualize_trades(strategy: PtResearchStrategy, results: PairResearchResult,
type="line", type="line",
x0=timeline_df['tstamp'].min(), x0=timeline_df['tstamp'].min(),
x1=timeline_df['tstamp'].max(), x1=timeline_df['tstamp'].max(),
y0=strategy.config_.get_value('dis-equilibrium_open_trshld'), y0=strategy.config_.get_value('model/disequilibrium/open_trshld'),
y1=strategy.config_.get_value('dis-equilibrium_open_trshld'), y1=strategy.config_.get_value('model/disequilibrium/open_trshld'),
line=dict(color="purple", width=2, dash="dot"), line=dict(color="purple", width=2, dash="dot"),
opacity=0.7, opacity=0.7,
row=1, col=1 row=1, col=1
@ -116,8 +116,8 @@ def visualize_trades(strategy: PtResearchStrategy, results: PairResearchResult,
type="line", type="line",
x0=timeline_df['tstamp'].min(), x0=timeline_df['tstamp'].min(),
x1=timeline_df['tstamp'].max(), x1=timeline_df['tstamp'].max(),
y0=-strategy.config_.get_value('dis-equilibrium_open_trshld'), y0=-strategy.config_.get_value('model/disequilibrium/open_trshld'),
y1=-strategy.config_.get_value('dis-equilibrium_open_trshld'), y1=-strategy.config_.get_value('model/disequilibrium/open_trshld'),
line=dict(color="purple", width=2, dash="dot"), line=dict(color="purple", width=2, dash="dot"),
opacity=0.7, opacity=0.7,
row=1, col=1 row=1, col=1
@ -127,8 +127,8 @@ def visualize_trades(strategy: PtResearchStrategy, results: PairResearchResult,
type="line", type="line",
x0=timeline_df['tstamp'].min(), x0=timeline_df['tstamp'].min(),
x1=timeline_df['tstamp'].max(), x1=timeline_df['tstamp'].max(),
y0=strategy.config_.get_value('dis-equilibrium_close_trshld'), y0=strategy.config_.get_value('model/disequilibrium/close_trshld'),
y1=strategy.config_.get_value('dis-equilibrium_close_trshld'), y1=strategy.config_.get_value('model/disequilibrium/close_trshld'),
line=dict(color="brown", width=2, dash="dot"), line=dict(color="brown", width=2, dash="dot"),
opacity=0.7, opacity=0.7,
row=1, col=1 row=1, col=1
@ -138,8 +138,8 @@ def visualize_trades(strategy: PtResearchStrategy, results: PairResearchResult,
type="line", type="line",
x0=timeline_df['tstamp'].min(), x0=timeline_df['tstamp'].min(),
x1=timeline_df['tstamp'].max(), x1=timeline_df['tstamp'].max(),
y0=-strategy.config_.get_value('dis-equilibrium_close_trshld'), y0=-strategy.config_.get_value('model/disequilibrium/close_trshld'),
y1=-strategy.config_.get_value('dis-equilibrium_close_trshld'), y1=-strategy.config_.get_value('model/disequilibrium/close_trshld'),
line=dict(color="brown", width=2, dash="dot"), line=dict(color="brown", width=2, dash="dot"),
opacity=0.7, opacity=0.7,
row=1, col=1 row=1, col=1

1438
research/notebooks/pair_trading_test.ipynb
View File

File diff suppressed because one or more lines are too long