progress 0.0.4

.vscode/launch.json

@@ -30,12 +30,14 @@
             "env": {
                 "PYTHONPATH": "${workspaceFolder}/..",
                 "CONFIG_SERVICE": "cloud16.cvtt.vpn:6789",
-                "MODEL_CONFIG": "vecm"
+                "MODEL_CONFIG": "vecm",
+                // "CVTT_URL": "http://cvtt-tester-01.cvtt.vpn:23456",
+                "CVTT_URL": "http://dev-server-02.cvtt.vpn:23456",
             },
             "args": [
                 // "--config=${workspaceFolder}/configuration/pair_trader.cfg",
                 "--config=http://cloud16.cvtt.vpn:6789/apps/pairs_trading/pair_trader",
-                "--book_id=TEST_BOOK_20250818",
+                "--book_id=TSTBOOK_PT_20260113",
                 "--instrument_A=COINBASE_AT:PAIR-ADA-USD", 
                 "--instrument_B=COINBASE_AT:PAIR-SOL-USD",
             ],

VERSION

@@ -0,0 +1 @@
+0.0.4

apps/pair_trader.py

@@ -141,12 +141,11 @@ class PairTrader(NamedObject):
             )
 
     async def _on_md_summary(self, history: List[MdTradesAggregate], exch_inst: ExchangeInstrument) -> None:
-        # URGENT before calling stragegy, make sure that **BOTH** instruments market data is combined.
-        Log.info(f"DEBUG got {exch_inst.details_short()} data")
+        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.time_ns_)
+            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_ = {}

build.sh

@@ -0,0 +1,183 @@
+#!/usr/bin/env bash
+
+# ---------------- Settings
+
+repo=git@cloud21.cvtt.vpn:/works/git/cvtt2/research/pairs_trading.git
+
+dist_root=/home/cvttdist/software/cvtt2
+dist_user=cvttdist
+dist_host="cloud21.cvtt.vpn"
+dist_ssh_port="22"
+
+dist_locations="cloud21.cvtt.vpn:22 hs01.cvtt.vpn:22"
+version_file="VERSION"
+
+prj=pairs_trading
+brnch=master
+interactive=N
+
+# ---------------- Settings
+
+# ---------------- cmdline
+
+usage() {
+  echo "Usage: $0 [-b <branch (master)> -i (interactive)"
+  exit 1
+}
+
+while getopts "b:i" opt; do
+  case ${opt} in
+    b )
+      brnch=$OPTARG
+      ;;
+    i )
+      interactive=Y
+      ;;
+    \? )
+      echo "Invalid option: -$OPTARG" >&2
+      usage
+      ;;
+    : )
+      echo "Option -$OPTARG requires an argument." >&2
+      usage
+      ;;
+  esac
+done
+# ---------------- cmdline
+
+confirm() {
+    if [ "${interactive}" == "Y" ]; then
+        echo "--------------------------------"
+        echo -n "Press <Enter> to continue" && read
+    fi
+}
+
+
+if [ "${interactive}" == "Y" ]; then
+    echo -n "Enter project [${prj}]: "
+    read project
+    if [ "${project}" == "" ]
+    then
+        project=${prj}
+    fi
+else
+    project=${prj}
+fi
+
+# repo=${git_repo_arr[${project}]}
+if [ -z ${repo} ]; then
+    echo "ERROR: Project repository for ${project} not found"
+    exit -1
+fi
+echo "Project repo: ${repo}"
+
+if [ "${interactive}" == "Y" ]; then
+    echo -n "Enter branch to build release from [${brnch}]: "
+    read branch
+    if [ "${branch}" == "" ]
+    then
+        branch=${brnch}
+    fi
+else
+    branch=${brnch}
+fi
+
+tmp_dir=$(mktemp -d)
+function cleanup {
+    cd ${HOME}
+    rm -rf ${tmp_dir}
+}
+trap cleanup EXIT
+
+
+prj_dir="${tmp_dir}/${prj}"
+
+cmd_arr=()
+Cmd="git clone ${repo} ${prj_dir}"
+cmd_arr+=("${Cmd}")
+
+Cmd="cd ${prj_dir}"
+cmd_arr+=("${Cmd}")
+
+if [ "${interactive}" == "Y" ]; then
+    echo "------------------------------------"
+    echo "The following commands will execute:"
+    echo "------------------------------------"
+    for cmd in "${cmd_arr[@]}"
+    do
+        echo ${cmd}
+    done
+fi
+
+confirm
+
+for cmd in "${cmd_arr[@]}"
+do
+    echo ${cmd} && eval ${cmd}
+done
+
+Cmd="git checkout ${branch}"
+echo ${Cmd} && eval ${Cmd}
+if [ "${?}" != "0" ]; then
+    echo "ERROR: Branch ${branch} is not found"
+    cd ${HOME} && rm -rf ${tmp_dir}
+    exit -1
+fi
+
+
+release_version=$(cat ${version_file} | awk -F',' '{print $1}')
+whats_new=$(cat ${version_file} | awk -F',' '{print $2}')
+
+
+echo "--------------------------------"
+echo "Version file: ${version_file}"
+echo "Release version: ${release_version}"
+
+confirm
+
+version_tag="v${release_version}"
+version_comment="'${version_tag} ${project} ${branch} $(date +%Y-%m-%d)\n${whats_new}'"
+
+cmd_arr=()
+Cmd="git tag -a ${version_tag} -m ${version_comment}"
+cmd_arr+=("${Cmd}")
+
+Cmd="git push origin --tags"
+cmd_arr+=("${Cmd}")
+
+Cmd="rm -rf .git"
+cmd_arr+=("${Cmd}")
+
+SourceLoc=../${project}
+
+dist_path="${dist_root}/${project}/${release_version}"
+
+for dist_loc in ${dist_locations}; do
+    dhp=(${dist_loc//:/ })
+    dist_host=${dhp[0]}
+    dist_port=${dhp[1]}
+    Cmd="rsync -avzh"
+    Cmd="${Cmd} --rsync-path=\"mkdir -p ${dist_path}"
+    Cmd="${Cmd} && rsync\" -e \"ssh -p ${dist_ssh_port}\""
+    Cmd="${Cmd} $SourceLoc ${dist_user}@${dist_host}:${dist_path}/"
+    cmd_arr+=("${Cmd}")
+done
+
+if [ "${interactive}" == "Y" ]; then
+    echo "------------------------------------"
+    echo "The following commands will execute:"
+    echo "------------------------------------"
+    for cmd in "${cmd_arr[@]}"
+    do
+        echo ${cmd}
+    done
+fi
+
+confirm
+
+for cmd in "${cmd_arr[@]}"
+do
+    pwd && echo ${cmd} && eval ${cmd}
+done
+
+echo "$0 Done ${project} ${release_version}"

lib/live/mkt_data_client.py

@@ -163,6 +163,7 @@ class MdSummaryCollector(NamedObject):
             )
             return None
         res = MdSummary.from_REST_response(response=response)
+        Log.info(f"DEBUG *** {self.exch_inst_.base_asset_id_}: {res[-1].tstamp_}")
         return None if len(res) == 0 else res[-1]
 
     def is_empty(self) -> bool:
@@ -195,15 +196,16 @@ class MdSummaryCollector(NamedObject):
         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:
+        ALLOW_LAG_SEC = 1
         curr_sec = int(current_seconds())
-        return (curr_sec - curr_sec % self.interval_sec_) + self.interval_sec_ + 2
+        return (curr_sec - curr_sec % self.interval_sec_) + self.interval_sec_ + ALLOW_LAG_SEC
 
     async def _load_new(self) -> None:
 
         last: Optional[MdSummary] = self.get_last()
         if not last:
             Log.warning(f"{self.fname()}: did not get last update")
-        elif not self.is_empty() and last.ts_ns_ <= self.history_[-1].time_ns_:
+        elif not self.is_empty() and last.ts_ns_ <= self.history_[-1].aggr_time_ns_:
             Log.info(
                 f"{self.fname()}: Received {last}. Already Have: {self.history_[-1]}"
             )

lib/live/rest_client.py

@@ -1,18 +1,12 @@
 from __future__ import annotations
 
-import asyncio
-from typing import Callable, Dict, Any, List, Optional
+from typing import Dict
 import time
 
 import requests
 
 from cvttpy_tools.base import NamedObject
-from cvttpy_tools.logger import Log
-from cvttpy_tools.config import Config
-from cvttpy_tools.timer import Timer
 
-from cvttpy_tools.timeutils import NanoPerSec, NanosT, current_nanoseconds, current_seconds
-from cvttpy_trading.trading.mkt_data.historical_md import HistMdBar
 
 
 class RESTSender(NamedObject):

lib/pt_strategy/live/live_strategy.py

@@ -9,7 +9,7 @@ from cvttpy_tools.base import NamedObject
 from cvttpy_tools.app import App
 from cvttpy_tools.config import Config
 from cvttpy_tools.settings.cvtt_types import IntervalSecT
-from cvttpy_tools.timeutils import SecPerHour, current_nanoseconds, NanoPerSec
+from cvttpy_tools.timeutils import NanosT, SecPerHour, current_nanoseconds, NanoPerSec, format_nanos_utc
 from cvttpy_tools.logger import Log
 
 # ---
@@ -42,14 +42,14 @@ class PtLiveStrategy(NamedObject):
     # for presentation: history of prediction values and trading signals
     predictions_df_: pd.DataFrame
     trading_signals_df_: pd.DataFrame
+    allowed_md_lag_sec_: int
+
 
     def __init__(
         self,
         config: Config,
         pairs_trader: PairTrader,
     ):
-        # import copy
-        # self.config_ = Config(json_src=copy.deepcopy(config.data()))
         self.config_ = config
 
         self.pairs_trader_ = pairs_trader
@@ -83,6 +83,8 @@ class PtLiveStrategy(NamedObject):
         )
         assert self.history_depth_sec_ > 0, "history_depth_hours cannot be 0"
 
+        self.allowed_md_lag_sec_ = self.config_.get_value("allowed_md_lag_sec", 3)
+
         await self.pairs_trader_.subscribe_md()
 
         self.open_threshold_ = self.config_.get_value(
@@ -132,17 +134,33 @@ class PtLiveStrategy(NamedObject):
             await self._send_trading_instructions(trading_instructions)
 
     def _is_md_actual(self, hist_aggr: List[MdTradesAggregate]) -> bool:
-        curr_ns = current_nanoseconds()
-        LAG_THRESHOLD = 5 * NanoPerSec
-        
         if len(hist_aggr) == 0:
             Log.warning(f"{self.fname()} list of aggregates IS EMPTY")
             return False
+
+        curr_ns = current_nanoseconds()
+        
         # MAYBE check market data length
-        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")
+        
+        # at 18:05:01 we should see data for 18:04:00
+        lag_sec = (curr_ns - hist_aggr[-1].aggr_time_ns_) / NanoPerSec - self.interval_sec()
+        if lag_sec > self.allowed_md_lag_sec_:
+            Log.warning(
+                f"{self.fname()} {hist_aggr[-1].exch_inst_.details_short()}"
+                f" Lagging {int(lag_sec)} > {self.allowed_md_lag_sec_} seconds:"
+                f"\n{len(hist_aggr)} records"
+                f"\n{hist_aggr[-1].exch_inst_.base_asset_id_}: {hist_aggr[-1].tstamp()}"
+                f"\n{hist_aggr[-2].exch_inst_.base_asset_id_}: {hist_aggr[-2].tstamp()}"
+            )
             return False
+        else:
+            Log.info(
+                f"{self.fname()} {hist_aggr[-1].exch_inst_.details_short()}"
+                f" Lag {int(lag_sec)} <= {self.allowed_md_lag_sec_} seconds"
+                f"\n{len(hist_aggr)} records"
+                f"\n{hist_aggr[-1].exch_inst_.base_asset_id_}: {hist_aggr[-1].tstamp()}"
+                f"\n{hist_aggr[-2].exch_inst_.base_asset_id_}: {hist_aggr[-2].tstamp()}"
+            )
         return True
 
     def _create_md_df(self, hist_aggr: List[MdTradesAggregate]) -> pd.DataFrame:
@@ -163,8 +181,8 @@ class PtLiveStrategy(NamedObject):
             rows.append(
                 {
                     # convert nanoseconds → tz-aware pandas timestamp
-                    "tstamp": pd.to_datetime(aggr.time_ns_, unit="ns", utc=True),
-                    "time_ns": aggr.time_ns_,
+                    "tstamp": pd.to_datetime(aggr.aggr_time_ns_, unit="ns", utc=True),
+                    "time_ns": aggr.aggr_time_ns_,
                     "symbol": exch_inst.instrument_id().split("-", 1)[1],
                     "exchange_id": exch_inst.exchange_id_,
                     "instrument_id": exch_inst.instrument_id(),
@@ -270,6 +288,7 @@ class PtLiveStrategy(NamedObject):
             issued_ts_ns=current_nanoseconds(),
             data=TargetPositionSignal(
                 strength=side_a * self._strength(scaled_disequilibrium),
+                exchange_id=pair.get_instrument_a().exchange_id_,
                 base_asset=pair.get_instrument_a().base_asset_id_,
                 quote_asset=pair.get_instrument_a().quote_asset_id_,
                 user_data={}
@@ -284,6 +303,7 @@ class PtLiveStrategy(NamedObject):
             issued_ts_ns=current_nanoseconds(),
             data=TargetPositionSignal(
                 strength=side_b * self._strength(scaled_disequilibrium),
+                exchange_id=pair.get_instrument_b().exchange_id_,
                 base_asset=pair.get_instrument_b().base_asset_id_,
                 quote_asset=pair.get_instrument_b().quote_asset_id_,
                 user_data={}
@@ -304,6 +324,7 @@ class PtLiveStrategy(NamedObject):
             issued_ts_ns=current_nanoseconds(),
             data=TargetPositionSignal(
                 strength=0,
+                exchange_id=pair.get_instrument_a().exchange_id_,
                 base_asset=pair.get_instrument_a().base_asset_id_,
                 quote_asset=pair.get_instrument_a().quote_asset_id_,
                 user_data={}
@@ -318,6 +339,7 @@ class PtLiveStrategy(NamedObject):
             issued_ts_ns=current_nanoseconds(),
             data=TargetPositionSignal(
                 strength=0,
+                exchange_id=pair.get_instrument_b().exchange_id_,
                 base_asset=pair.get_instrument_b().base_asset_id_,
                 quote_asset=pair.get_instrument_b().quote_asset_id_,
                 user_data={}