170 lines
6.4 KiB
Python
170 lines
6.4 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import Callable, Coroutine, Dict, 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.live.mkt_data_client import CvttRestMktDataClient
|
|
|
|
'''
|
|
config http://cloud16.cvtt.vpn/apps/pairs_trading
|
|
'''
|
|
|
|
HistMdCbT = Callable[[List[MdTradesAggregate]], Coroutine]
|
|
UpdateMdCbT = Callable[[MdTradesAggregate], Coroutine]
|
|
|
|
class PairTrader(NamedObject):
|
|
config_: CvttAppConfig
|
|
instruments_: List[ExchangeInstrument]
|
|
book_id_: BookIdT
|
|
|
|
live_strategy_: "PtLiveStrategy" #type: ignore
|
|
ti_sender_: "TradingInstructionsSender" #type: ignore
|
|
pricer_client_: CvttRestMktDataClient
|
|
rest_service_: RestService
|
|
|
|
latest_history_: Dict[ExchangeInstrument, List[MdTradesAggregate]]
|
|
|
|
def __init__(self) -> None:
|
|
self.instruments_ = []
|
|
self.latest_history_ = {}
|
|
|
|
App.instance().add_cmdline_arg(
|
|
"--instrument_A",
|
|
type=str,
|
|
required=True,
|
|
help=(
|
|
" Instrument A in pair (e.g., COINBASE_AT:PAIR-BTC-USD)"
|
|
),
|
|
)
|
|
App.instance().add_cmdline_arg(
|
|
"--instrument_B",
|
|
type=str,
|
|
required=True,
|
|
help=(
|
|
" Instrument B in pair (e.g., COINBASE_AT:PAIR-ETH-USD)"
|
|
),
|
|
)
|
|
|
|
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_list: List[str] = []
|
|
instr_str = App.instance().get_argument("instrument_A", "")
|
|
assert instr_str != "", "Missing insrument A"
|
|
instr_list.append(instr_str)
|
|
instr_str = App.instance().get_argument("instrument_B", "")
|
|
assert instr_str != "", "Missing insrument B"
|
|
instr_list.append(instr_str)
|
|
|
|
for instr in instr_list:
|
|
instr_parts = instr.split(":")
|
|
if len(instr_parts) != 2:
|
|
raise ValueError(f"Invalid pair format: {instr}")
|
|
exch_acct = instr_parts[0]
|
|
instrument_id = 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 -------
|
|
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(
|
|
config=strategy_config,
|
|
pairs_trader=self,
|
|
)
|
|
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 -------
|
|
self.pricer_client_ = CvttRestMktDataClient(config=self.config_)
|
|
Log.info(f"{self.fname()} MD client created: {self.pricer_client_}")
|
|
|
|
# ------- CREATE TRADER CLIENT -------
|
|
from pairs_trading.lib.live.ti_sender import TradingInstructionsSender
|
|
self.ti_sender_ = TradingInstructionsSender(config=self.config_, pairs_trader=self)
|
|
Log.info(f"{self.fname()} TI sender 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:
|
|
from functools import partial
|
|
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=partial(self._on_md_summary, exch_inst=exch_inst)
|
|
)
|
|
|
|
async def _on_md_summary(self, history: List[MdTradesAggregate], exch_inst: ExchangeInstrument) -> None:
|
|
Log.info(f"{self.fname()}: 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.aggr_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:
|
|
# TODO choose pair
|
|
# TODO confirm chosen pair (after selection is implemented)
|
|
return web.Response() # TODO API request handler implementation
|
|
|
|
|
|
async def run(self) -> None:
|
|
Log.info(f"{self.fname()} ...")
|
|
while True:
|
|
await asyncio.sleep(0.1)
|
|
pass
|
|
|
|
if __name__ == "__main__":
|
|
App()
|
|
CvttAppConfig()
|
|
PairTrader()
|
|
App.instance().run()
|