147 lines
5.3 KiB
Python
147 lines
5.3 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Callable, Coroutine, List
|
|
import aiohttp.web as web
|
|
|
|
from cvttpy_tools.app import App
|
|
from cvttpy_tools.config import Config
|
|
from cvttpy_tools.base import NamedObject
|
|
from cvttpy_tools.config import CvttAppConfig
|
|
from cvttpy_tools.logger import Log
|
|
from cvttpy_tools.settings.cvtt_types import BookIdT
|
|
from cvttpy_tools.web.rest_service import RestService
|
|
|
|
# ---
|
|
from cvttpy_trading.trading.instrument import ExchangeInstrument
|
|
from cvttpy_trading.trading.mkt_data.md_summary import MdTradesAggregate
|
|
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.ti_sender import TradingInstructionsSender
|
|
|
|
'''
|
|
config http://cloud16.cvtt.vpn/apps/pairs_trading
|
|
'''
|
|
|
|
HistMdCbT = Callable[[List[MdTradesAggregate]], Coroutine]
|
|
UpdateMdCbT = Callable[[MdTradesAggregate], Coroutine]
|
|
|
|
class PairsTrader(NamedObject):
|
|
config_: CvttAppConfig
|
|
instruments_: List[ExchangeInstrument]
|
|
book_id_: BookIdT
|
|
|
|
live_strategy_: PtLiveStrategy
|
|
pricer_client_: CvttRestMktDataClient
|
|
ti_sender_: TradingInstructionsSender
|
|
rest_service_: RestService
|
|
|
|
def __init__(self) -> None:
|
|
self.instruments_ = []
|
|
|
|
App.instance().add_cmdline_arg(
|
|
"--pair",
|
|
type=str,
|
|
required=True,
|
|
help=(
|
|
"Comma-separated pair of instrument symbols"
|
|
" with exchange config name"
|
|
" (e.g., PAIR-BTC-USD:BNBSPOT,PAIR-ETH-USD:BNBSPOT)"
|
|
),
|
|
)
|
|
|
|
App.instance().add_cmdline_arg(
|
|
"--book_id",
|
|
type=str,
|
|
required=True,
|
|
help="Book ID"
|
|
)
|
|
App.instance().add_call(App.Stage.Config, self._on_config())
|
|
App.instance().add_call(App.Stage.Run, self.run())
|
|
|
|
async def _on_config(self) -> None:
|
|
self.config_ = CvttAppConfig.instance()
|
|
self.book_id_ = App.instance().get_argument(name="book_id")
|
|
|
|
# ------- PARSE INSTRUMENTS -------
|
|
instr_str = App.instance().get_argument("pair", "")
|
|
if not instr_str:
|
|
raise ValueError("Pair is required")
|
|
instr_list = instr_str.split(",")
|
|
|
|
assert len(instr_list) == 2, "Only two instruments are supported"
|
|
|
|
for instr in instr_list:
|
|
instr_parts = instr.split(":")
|
|
if len(instr_parts) != 2:
|
|
raise ValueError(f"Invalid pair format: {instr}")
|
|
instrument_id = instr_parts[0]
|
|
exch_acct = instr_parts[1]
|
|
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}"
|
|
exch_inst.user_data_["exch_acct"] = exch_acct
|
|
self.instruments_.append(exch_inst)
|
|
|
|
Log.info(f"{self.fname()} Instruments: {self.instruments_[0].details_short()} <==> {self.instruments_[1].details_short()}")
|
|
|
|
# ------- CREATE STRATEGY -------
|
|
strategy_config = self.config_.get_subconfig("strategy_config", Config({}))
|
|
self.live_strategy_ = PtLiveStrategy(
|
|
config=strategy_config,
|
|
instruments=self.instruments_,
|
|
pairs_trader=self
|
|
)
|
|
Log.info(f"{self.fname()} Strategy created: {self.live_strategy_}")
|
|
|
|
# # ------- CREATE PRICER CLIENT -------
|
|
self.pricer_client_ = CvttRestMktDataClient(config=self.config_)
|
|
Log.info(f"{self.fname()} MD client created: {self.pricer_client_}")
|
|
|
|
# ------- CREATE TRADER CLIENT -------
|
|
self.ti_sender_ = TradingInstructionsSender(config=self.config_, pairs_trader=self)
|
|
Log.info(f"{self.fname()} TI sebder created: {self.ti_sender_}")
|
|
|
|
# # ------- CREATE REST SERVER -------
|
|
self.rest_service_ = RestService(
|
|
config_key=f"/api/REST"
|
|
)
|
|
|
|
# --- Strategy Handlers
|
|
self.rest_service_.add_handler(
|
|
method="POST",
|
|
url="/api/strategy",
|
|
handler=self._on_api_request,
|
|
)
|
|
|
|
async def subscribe_md(self) -> None:
|
|
for exch_inst in self.instruments_:
|
|
exch_acct = exch_inst.user_data_.get("exch_acct", "?exch_acct?")
|
|
instrument_id = exch_inst.instrument_id()
|
|
|
|
await self.pricer_client_.add_subscription(
|
|
exch_acct=exch_acct,
|
|
instrument_id=instrument_id,
|
|
interval_sec=self.live_strategy_.interval_sec(),
|
|
history_depth_sec=self.live_strategy_.history_depth_sec(),
|
|
callback=self._on_md_summary
|
|
)
|
|
|
|
async def _on_md_summary(self, history: List[MdTradesAggregate]) -> None:
|
|
# Snapshot or update?
|
|
await self.live_strategy_.on_mkt_data_hist_snapshot(hist_aggr=history)
|
|
|
|
async def _on_api_request(self, request: web.Request) -> web.Response:
|
|
return web.Response() # TODO API request handler implementation
|
|
|
|
async def run(self) -> None:
|
|
Log.info(f"{self.fname()} ...")
|
|
pass
|
|
|
|
if __name__ == "__main__":
|
|
App()
|
|
CvttAppConfig()
|
|
PairsTrader()
|
|
App.instance().run()
|