Compare commits

..

2 Commits

Author SHA1 Message Date
Oleg Sheynin 8a207a13a0 . 2026-01-30 20:21:44 +00:00
Oleg Sheynin 38c3c73021 stats history for pair selector 2026-01-28 23:27:52 +00:00
58 changed files with 1167 additions and 5114 deletions
+4 -17
View File
@@ -3,22 +3,9 @@ __pycache__/
__OLD__/
.specstory/
.history/
.vscode/
*.py[cod]
.ipynb_checkpoints/
.pytest_cache/
# Local environments
.venv/
venv/
# Local test data and generated analysis results
data/*
!data/.gitkeep
results/*
!results/.gitkeep
.cursorindexingignore
data
cvttpy
tmp/
# SpecStory explanation file
.specstory/.what-is-this.md
results/
+1
View File
@@ -0,0 +1 @@
PYTHONPATH=/home/oleg/develop
+177
View File
@@ -0,0 +1,177 @@
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Python Debugger: Current File",
"type": "debugpy",
"request": "launch",
"python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
"program": "${file}",
"console": "integratedTerminal",
"env": {
"PYTHONPATH": "${workspaceFolder}/lib:${workspaceFolder}/.."
},
},
{
"name": "-------- Live Pair Trading --------",
},
{
"name": "PAIR TRADER",
"type": "debugpy",
"request": "launch",
"python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
"program": "${workspaceFolder}/apps/pair_trader.py",
"console": "integratedTerminal",
"env": {
"PYTHONPATH": "${workspaceFolder}/..",
"CONFIG_SERVICE": "cloud16.cvtt.vpn:6789",
"CVTT_URL": "http://cvtt-tester-01.cvtt.vpn:23456",
"MODEL_CONFIG": "vecm",
"PAIR_TRADER_REST_PORT": "54320"
},
"args": [
// "--config=${workspaceFolder}/configuration/pair_trader.cfg",
"--config=http://cloud16.cvtt.vpn:6789/apps/pairs_trading/pair_trader",
"--book_id=TSTBOOK_PT_20260113",
"--instrument_A=COINBASE_AT:PAIR-ADA-USD",
"--instrument_B=COINBASE_AT:PAIR-SOL-USD",
],
},
{
"name": "PAIR SELECTOR",
"type": "debugpy",
"request": "launch",
"python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
"program": "${workspaceFolder}/apps/pair_selector.py",
"console": "integratedTerminal",
"env": {
"PYTHONPATH": "${workspaceFolder}/..",
"CONFIG_SERVICE": "cloud16.cvtt.vpn:6789",
// "CVTT_URL": "http://cvtt-tester-01.cvtt.vpn:23456",
"CVTT_URL": "http://dev-server-02.cvtt.vpn:23456",
"PAIR_SELECTOR_REST_PORT": "44320"
},
"args": [
// "--config=${workspaceFolder}/configuration/pair_trader.cfg",
"--config=http://cloud16.cvtt.vpn:6789/apps/pairs_trading/pair_selector",
],
},
{
"name": "-------- VECM --------",
},
{
"name": "CRYPTO VECM BACKTEST (optimized)",
"type": "debugpy",
"request": "launch",
"python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
"program": "${workspaceFolder}/research/backtest.py",
"args": [
"--config=http://cloud16.cvtt.vpn:6789/apps/pairs_trading/backtest",
"--instruments=CRYPTO:BNBSPOT:PAIR-ADA-USDT,CRYPTO:BNBSPOT:PAIR-SOL-USDT",
"--date_pattern=20250911",
"--result_db=${workspaceFolder}/research/results/crypto/%T.vecm-opt.ADA-SOL.20250605.crypto_results.db",
],
"env": {
"PYTHONPATH": "${workspaceFolder}/..",
"CONFIG_SERVICE": "cloud16.cvtt.vpn:6789",
"MODEL_CONFIG": "vecm-opt"
},
"console": "integratedTerminal"
},
// {
// "name": "EQUITY VECM (rolling)",
// "type": "debugpy",
// "request": "launch",
// "python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
// "program": "${workspaceFolder}/research/backtest.py",
// "args": [
// "--config=${workspaceFolder}/configuration/vecm.cfg",
// "--instruments=COIN:EQUITY:ALPACA,MSTR:EQUITY:ALPACA",
// "--date_pattern=20250605",
// "--result_db=${workspaceFolder}/research/results/equity/%T.vecm.COIN-MSTR.20250605.equity_results.db",
// ],
// "env": {
// "PYTHONPATH": "${workspaceFolder}/lib"
// },
// "console": "integratedTerminal"
// },
// {
// "name": "EQUITY-CRYPTO VECM (rolling)",
// "type": "debugpy",
// "request": "launch",
// "python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
// "program": "${workspaceFolder}/research/backtest.py",
// "args": [
// "--config=${workspaceFolder}/configuration/vecm.cfg",
// "--instruments=COIN:EQUITY:ALPACA,BTC-USDT:CRYPTO:BNBSPOT",
// "--date_pattern=20250605",
// "--result_db=${workspaceFolder}/research/results/intermarket/%T.vecm.COIN-BTC.20250601.equity_results.db",
// ],
// "env": {
// "PYTHONPATH": "${workspaceFolder}/lib"
// },
// "console": "integratedTerminal"
// },
{
"name": "-------- B a t c h e s --------",
},
{
"name": "CRYPTO OLS Batch (rolling)",
"type": "debugpy",
"request": "launch",
"python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
"program": "${workspaceFolder}/research/backtest.py",
"args": [
"--config=${workspaceFolder}/configuration/ols.cfg",
"--instruments=ADA-USDT:CRYPTO:BNBSPOT,SOL-USDT:CRYPTO:BNBSPOT",
"--date_pattern=2025060*",
"--result_db=${workspaceFolder}/research/results/crypto/%T.ols.ADA-SOL.2025060-.crypto_results.db",
],
"env": {
"PYTHONPATH": "${workspaceFolder}/lib"
},
"console": "integratedTerminal"
},
{
"name": "CRYPTO VECM Batch (rolling)",
"type": "debugpy",
"request": "launch",
"python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
"program": "${workspaceFolder}/research/backtest.py",
"args": [
"--config=${workspaceFolder}/configuration/vecm.cfg",
"--instruments=ADA-USDT:CRYPTO:BNBSPOT,SOL-USDT:CRYPTO:BNBSPOT",
"--date_pattern=2025060*",
"--result_db=${workspaceFolder}/research/results/crypto/%T.vecm.ADA-SOL.2025060-.crypto_results.db",
],
"env": {
"PYTHONPATH": "${workspaceFolder}/lib"
},
"console": "integratedTerminal"
},
{
"name": "-------- Viz Test --------",
},
{
"name": "Viz Test",
"type": "debugpy",
"request": "launch",
"python": "/home/oleg/.pyenv/python3.12-venv/bin/python",
"program": "${workspaceFolder}/tests/viz_test.py",
"args": [
"--config=${workspaceFolder}/configuration/ols.cfg",
"--instruments=ADA-USDT:CRYPTO:BNBSPOT,SOL-USDT:CRYPTO:BNBSPOT",
"--date_pattern=20250605",
],
"env": {
"PYTHONPATH": "${workspaceFolder}/lib"
},
"console": "integratedTerminal"
}
]
}
+10
View File
@@ -0,0 +1,10 @@
{
"folders": [
{
"path": ".."
}
],
"settings": {
"workbench.colorTheme": "Dracula Theme"
}
}
+19
View File
@@ -0,0 +1,19 @@
{
"python.testing.pytestEnabled": true,
"python.testing.unittestEnabled": false,
"python.testing.pytestArgs": [
"unittests"
],
"python.testing.cwd": "${workspaceFolder}",
"python.testing.autoTestDiscoverOnSaveEnabled": true,
"python.defaultInterpreterPath": "/home/oleg/.pyenv/python3.12-venv/bin/python3",
"python.testing.pytestPath": "python3",
"python.analysis.extraPaths": [
"${workspaceFolder}",
"${workspaceFolder}/..",
"${workspaceFolder}/unittests"
],
"python.envFile": "${workspaceFolder}/.env",
"python.testing.debugPort": 3000,
"python.testing.promptToConfigure": false,
}
-156
View File
@@ -1,156 +0,0 @@
# Agent Instructions
## Repository purpose
This repository analyzes test results with Jupyter notebooks and Python or
Bash scripts. Inputs are commonly SQLite databases containing time-series data
and JSON columns, but analyses may use other test-result formats.
Ignore `__SAV__/`. It is unrelated legacy material, is not part of the active
project, and must not be read, edited, moved, or used as a source of conventions
unless the user explicitly requests it.
## Active layout
- `notebooks/`: exploratory and report-oriented Jupyter notebooks.
- `scripts/`: reusable Python and Bash analysis utilities.
- `data/`: local input data. Contents are ignored except for `.gitkeep`.
- `results/`: generated tables, figures, exports, and reports. Contents are
ignored except for `.gitkeep`.
- `requirements.txt`: Python dependencies needed to reproduce repository work.
Keep reusable logic in `scripts/` and use notebooks to orchestrate analysis,
explain decisions, and present results. Do not create a separate `analysis/`
tree.
## Python environment
The intended virtual environment is `~/.pyenv/python3.12-venv`.
```bash
source ~/.pyenv/python3.12-venv/bin/activate
python -m pip install -r requirements.txt
```
Agents may install packages in this environment when needed. Whenever a package
is installed for repository work, update `requirements.txt` in the same change
with a suitable direct dependency declaration. Use `python -m pip`, not bare
`pip`, in documented commands.
Do not create an in-repository virtual environment unless the user asks for
one.
## Data handling
- Treat files in `data/` as local, potentially large, and potentially
sensitive.
- Do not commit SQLite databases, raw test results, or generated results.
- Do not modify source data in place. Write transformed data and exports under
`results/`.
- Use parameterized SQL for values. Do not construct SQL by interpolating
untrusted data.
- Parse JSON columns defensively and preserve missing, malformed, and unexpected
values unless the analysis explicitly defines another policy.
- State assumptions about timestamps, time zones, ordering, units, and duplicate
observations in the notebook or script that relies on them.
- Avoid loading entire databases into memory when a filtered query or chunked
read is practical.
## Notebook conventions
- A notebook must run from a fresh kernel, top to bottom, without relying on
hidden interactive state.
- Set random seeds where nondeterminism affects results.
- Keep data paths relative to the repository root and avoid machine-specific
absolute paths.
- Move logic that is reused or substantial enough to test into `scripts/`.
- Clear cell outputs before committing notebooks. Never commit embedded source
data, credentials, or bulky generated output.
- Keep concise Markdown context near analyses: purpose, input assumptions,
method, and interpretation.
## Scripts
- Python scripts should expose reusable functions and use a guarded CLI entry
point when executable.
- Bash scripts must start with `#!/usr/bin/env bash` and use
`set -euo pipefail`.
- Prefer explicit CLI arguments over hard-coded paths or parameters.
- Fail with actionable error messages when required data, tables, columns, or
configuration are missing.
## Verification
Verification should be proportional to the change. At minimum:
- Run `pytest` for Python script changes.
- Add or update tests for reusable parsing, transformation, query, and
calculation logic.
- Execute changed notebooks from a fresh kernel with `nbmake`.
- Run changed Bash scripts against a safe fixture or exercise their
non-destructive validation/help path.
- Clear notebook outputs after execution and before committing.
Useful commands:
```bash
python -m pytest
python -m pytest --nbmake notebooks
jupyter nbconvert --ClearOutputPreprocessor.enabled=True --inplace path/to/notebook.ipynb
```
If verification cannot be run, report exactly what was skipped and why.
## Release rules
- Update `CHANGELOG.md` for every release with the release version, release
date, Git tag, and a concise summary of notable changes.
- Keep an `Unreleased` section at the top of `CHANGELOG.md` for changes that
have not been included in a tagged release yet.
- Move relevant entries from `Unreleased` into the dated release section when
creating a release, and leave `Unreleased` present for future changes.
- Use release headers in `YYYY-MM-DD vMAJOR.MINOR.PATCH` form.
- Use version numbers in `MAJOR.MINOR.PATCH` form. Start this repository at
`0.0.1`.
- Use Git tags in `vMAJOR.MINOR.PATCH` form, matching the changelog version
exactly. For example, version `0.0.1` must be tagged as `v0.0.1`.
- Create the Git tag only after the changelog and any release-related version
changes are complete.
- When the user requests creating a release, treat that as explicit permission
to commit the release changes, create the matching Git tag, and push both the
branch and tag.
- Do not push release commits or tags unless the user explicitly requests it.
## Mandatory background review
Changes to Python scripts, Bash scripts, or notebook code cells require approval
from a separate background reviewer agent before the implementing agent may
declare the work complete.
The implementing agent must:
1. Finish the implementation and run the relevant verification.
2. Ask a separate background agent to review the diff for correctness,
reproducibility, data safety, and test coverage.
3. Address every material finding, rerun affected checks, and request follow-up
review when the fix materially changes the code.
4. Report the reviewer outcome in the final response.
The reviewer must inspect the actual diff and relevant surrounding files; a
self-review does not satisfy this requirement. Documentation-only,
configuration-only, dependency-only, and ignore-rule-only changes do not
require background approval unless they also alter Python, Bash, or notebook
code cells.
If no background reviewer is available, complete all other work but do not
claim reviewer approval. End the handoff with the exact status:
`review pending`
## Change discipline
- Preserve user changes and avoid unrelated cleanup.
- Do not edit or commit generated files from `data/` or `results/`.
- Do not push or commit unless the user explicitly requests it. The `master`
branch being unprotected does not imply permission to push directly.
- Keep changes focused and explain any new assumptions or dependencies.
-71
View File
@@ -1,71 +0,0 @@
# Changelog
All notable changes to this project are documented in this file.
## Unreleased
No unreleased changes yet.
## 2026-07-29 v1.0.2
- Added a Panel application for single-day SPBT result analysis with result-file
selection, minimum TARGET-change input, pair TheoRet table, pair selector,
selected-pair execution table, and market/trade chart.
- Added a launcher script for the Panel application.
- Changed notebook and Panel pair analysis to use per-row Analyze actions from
the Pair TheoRet grid, deferring selected-pair calculations until clicked.
- Adjusted Panel sizing so key controls use compact widths and Pair TheoRet uses
content width with vertical scrolling instead of full-width paginated layout.
- Added a FastListTemplate shell to the Panel application for sidebar controls
and configurable app color accents.
- Made Plotly chart panes use all available horizontal space.
## 2026-07-28 v1.0.1
- Added the `spbt_day` notebook for interactive single-day backtest result
analysis, including SQLite result file selection from the local data
directory.
- Added selector-pair loading and dense ranking by `mr_score.final`, preserving
rows with invalid score JSON for inspection.
- Added theoretical return calculation for ranked pairs from
`trading_instructions`, including reusable helper functions and tests.
- Added a Plotly histogram for visual analysis of total theoretical return by
pair.
- Moved notebook support code into reusable `scripts/spbt_day.py` helpers.
- Adjusted notebook table outputs to show all relevant rows and reduce
redundant intermediate displays.
- Added an alphabetically sorted pair selector for individual pair analysis.
- Added selected-pair theoretical execution tables and aligned TheoRet
calculations with target-delta trade generation.
- Added per-asset `strength` values to selected-pair theoretical execution
tables.
- Corrected theoretical execution size to use
`10000 * strength / reference_price`.
- Removed `:USD` quote suffixes from displayed pair names in notebook tables,
chart hovers, and the pair selector dropdown while preserving full internal
pair keys for calculations.
- Added `num_trades` to pair TheoRet summaries, counting asset-level theoretical
trades from effective `TARGET` and `CLOSE` instructions.
- Added sortable interactive grids for the pair TheoRet and selected-pair
theoretical execution tables.
- Styled interactive dataframe grids with black text on white backgrounds for
readability across notebook themes.
- Added a selected-pair Plotly chart that overlays theoretical BUY/SELL
executions on relative 1-minute market close data for both instruments.
- Anchored the selected-pair market chart at trading-day midnight and normalized
relative prices to each instrument's close at that timestamp.
- Added a `min_pctg_change` threshold for ranked pair TheoRet calculations to
skip small target-strength changes after a position is acquired.
- Added a notebook input field for the minimum TARGET strength-change threshold.
## 2026-07-25 v0.0.9
- Added contributing guidance and Python dependency declarations.
- Added placeholder files for active project directories.
- Updated ignore rules for local data, generated results, caches, and local
environments.
- Documented unreleased changelog handling and release push behavior.
## 2026-07-25 v0.0.1
- Established the initial repository structure and project guidance.
-54
View File
@@ -1,54 +0,0 @@
# Contributing
## Setup
Use the shared Python 3.12 virtual environment:
```bash
source ~/.pyenv/python3.12-venv/bin/activate
python -m pip install -r requirements.txt
```
If you install another package for repository work, add its direct dependency
to `requirements.txt`.
## Repository layout
- Put notebooks in `notebooks/`.
- Put reusable Python and Bash utilities in `scripts/`.
- Put local input files in `data/`.
- Put generated artifacts in `results/`.
The contents of `data/` and `results/` are ignored. Do not force-add test
databases, raw test results, generated exports, or notebook outputs.
`__SAV__/` is unrelated legacy material and is outside the active project.
## Working with notebooks
Notebooks must execute from top to bottom in a fresh kernel. Use relative paths,
document data assumptions, and move reusable logic into tested scripts.
Before handing off a change:
```bash
python -m pytest
python -m pytest --nbmake notebooks
jupyter nbconvert --ClearOutputPreprocessor.enabled=True --inplace path/to/notebook.ipynb
```
Run only the checks relevant to the files present in the repository, and report
anything that could not be run.
## Review requirement
Python scripts, Bash scripts, and notebook code-cell changes require review and
approval by a separate background agent. Address material findings and rerun
affected checks before completion. If a reviewer is unavailable, the change may
be handed off only with the status `review pending`.
Documentation, dependency declarations, and ignore rules do not require this
background review when no Python, Bash, or notebook code cells changed.
The `master` branch is not protected. That does not remove the review
requirement or authorize an agent to commit or push without an explicit request.
View File
+1
View File
@@ -0,0 +1 @@
0.0.6
-2
View File
@@ -1,2 +0,0 @@
## 2026-02-09 (v0.0.9)
- related to the changes made in *cvttpy_tools 1.4.7*
-1
View File
@@ -1 +0,0 @@
0.0.9
-937
View File
@@ -1,937 +0,0 @@
from __future__ import annotations
import asyncio
import os
import sqlite3
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Sequence, Set, Tuple, Union
from aiohttp import web
import numpy as np
import pandas as pd
from statsmodels.tsa.stattools import adfuller, coint # type: ignore
from statsmodels.tsa.vector_ar.vecm import coint_johansen # type: ignore
from cvttpy_tools.base.app import App
from cvttpy_tools.base.base import NamedObject
from cvttpy_tools.base.config import Config, CvttAppConfig
from cvttpy_tools.base.logger import Log
from cvttpy_tools.base.timeutils import NanoPerSec, SecPerHour, current_nanoseconds
from cvttpy_tools.comm.web.rest_service import RestService
from cvttpy_trading.trading.exchange_config import ExchangeAccounts
from cvttpy_trading.trading.instrument import ExchangeInstrument
from cvttpy_trading.trading.mkt_data.md_summary import MdTradesAggregate, MdSummary
from pairs_trading.apps.pair_selector.renderer import HtmlRenderer
from pairs_trading.lib.live.rest import RESTSender
@dataclass
class BacktestAggregate:
aggr_time_ns_: int
num_trades_: Optional[int]
@dataclass
class InstrumentQuality(NamedObject):
instrument_: ExchangeInstrument
record_count_: int
latest_tstamp_: Optional[pd.Timestamp]
status_: str
reason_: str
@dataclass
class PairStats(NamedObject):
pair_name_: str
instrument_a_: ExchangeInstrument
instrument_b_: ExchangeInstrument
pvalue_eg_: Optional[float]
pvalue_adf_: Optional[float]
pvalue_j_: Optional[float]
trace_stat_j_: Optional[float]
rank_eg_: int = 0
rank_adf_: int = 0
rank_j_: int = 0
composite_rank_: int = 0
def as_dict(self) -> Dict[str, Any]:
return {
"exchange_a": self.instrument_a_.exchange_id_,
"exchange_b": self.instrument_b_.exchange_id_,
"pair_name": self.pair_name_,
"instrument_a": self.instrument_a_.instrument_id(),
"instrument_b": self.instrument_b_.instrument_id(),
"pvalue_eg": self.pvalue_eg_,
"pvalue_adf": self.pvalue_adf_,
"pvalue_j": self.pvalue_j_,
"trace_stat_j": self.trace_stat_j_,
"rank_eg": self.rank_eg_,
"rank_adf": self.rank_adf_,
"rank_j": self.rank_j_,
"composite_rank": self.composite_rank_,
}
def _extract_price_from_fields(
price_field: str,
inst: ExchangeInstrument,
open: Optional[float],
high: Optional[float],
low: Optional[float],
close: Optional[float],
vwap: Optional[float],
) -> float:
field_map = {
"open": open,
"high": high,
"low": low,
"close": close,
"vwap": vwap,
}
raw = field_map.get(price_field, close)
if raw is None:
raw = 0.0
return inst.get_price(raw)
class DataFetcher(NamedObject):
sender_: RESTSender
interval_sec_: int
history_depth_sec_: int
def __init__(
self,
base_url: str,
interval_sec: int,
history_depth_sec: int,
) -> None:
self.sender_ = RESTSender(base_url=base_url)
self.interval_sec_ = interval_sec
self.history_depth_sec_ = history_depth_sec
def fetch(
self, exch_acct: str, inst: ExchangeInstrument
) -> List[MdTradesAggregate]:
rqst_data = {
"exch_acct": exch_acct,
"instrument_id": inst.instrument_id(),
"interval_sec": self.interval_sec_,
"history_depth_sec": self.history_depth_sec_,
}
response = self.sender_.send_post(endpoint="md_summary", post_body=rqst_data)
if response.status_code not in (200, 201):
Log.error(
f"{self.fname()}: error {response.status_code} for {inst.details_short()}: {response.text}"
)
return []
mdsums: List[MdSummary] = MdSummary.from_REST_response(response=response)
return [
mdsum.create_md_trades_aggregate(
exch_acct=exch_acct, exch_inst=inst, interval_sec=self.interval_sec_
)
for mdsum in mdsums
]
AggregateLike = Union[MdTradesAggregate, BacktestAggregate]
class QualityChecker(NamedObject):
interval_sec_: int
def __init__(self, interval_sec: int) -> None:
self.interval_sec_ = interval_sec
def evaluate(
self,
inst: ExchangeInstrument,
aggr: Sequence[AggregateLike],
now_ts: Optional[pd.Timestamp] = None,
) -> InstrumentQuality:
if len(aggr) == 0:
return InstrumentQuality(
instrument_=inst,
record_count_=0,
latest_tstamp_=None,
status_="FAIL",
reason_="no records",
)
aggr_sorted = sorted(aggr, key=lambda a: a.aggr_time_ns_)
latest_ts = pd.to_datetime(aggr_sorted[-1].aggr_time_ns_, unit="ns", utc=True)
now_ts = now_ts or pd.Timestamp.utcnow()
recency_cutoff = now_ts - pd.Timedelta(seconds=2 * self.interval_sec_)
if latest_ts <= recency_cutoff:
return InstrumentQuality(
instrument_=inst,
record_count_=len(aggr_sorted),
latest_tstamp_=latest_ts,
status_="FAIL",
reason_=f"stale: latest {latest_ts} <= cutoff {recency_cutoff}",
)
gaps_ok, reason = self._check_gaps(aggr_sorted)
status = "PASS" if gaps_ok else "FAIL"
return InstrumentQuality(
instrument_=inst,
record_count_=len(aggr_sorted),
latest_tstamp_=latest_ts,
status_=status,
reason_=reason,
)
def _check_gaps(self, aggr: Sequence[AggregateLike]) -> Tuple[bool, str]:
NUM_TRADES_THRESHOLD = 50
if len(aggr) < 2:
return True, "ok"
interval_ns = self.interval_sec_ * NanoPerSec
for idx in range(1, len(aggr)):
prev = aggr[idx - 1]
curr = aggr[idx]
delta = curr.aggr_time_ns_ - prev.aggr_time_ns_
missing_intervals = int(delta // interval_ns) - 1
if missing_intervals <= 0:
continue
prev_nt = prev.num_trades_
next_nt = curr.num_trades_
estimate = self._approximate_num_trades(prev_nt, next_nt)
if estimate > NUM_TRADES_THRESHOLD:
return False, (
f"gap of {missing_intervals} interval(s), est num_trades={estimate} > {NUM_TRADES_THRESHOLD}"
)
return True, "ok"
@staticmethod
def _approximate_num_trades(prev_nt: Optional[int], next_nt: Optional[int]) -> float:
if prev_nt is None and next_nt is None:
return 0.0
if prev_nt is None:
return float(next_nt or 0)
if next_nt is None:
return float(prev_nt)
return (prev_nt + next_nt) / 2.0
class PairAnalyzer(NamedObject):
price_field_: str
interval_sec_: int
def __init__(self, price_field: str, interval_sec: int) -> None:
self.price_field_ = price_field
self.interval_sec_ = interval_sec
def analyze(
self, series: Dict[ExchangeInstrument, pd.DataFrame]
) -> Dict[str, PairStats]:
instruments = list(series.keys())
results: Dict[str, PairStats] = {}
for i in range(len(instruments)):
for j in range(i + 1, len(instruments)):
inst_a, inst_b, pair_name = self._normalized_pair(
instruments[i], instruments[j]
)
df_a = series[inst_a][["tstamp", "price"]].rename(
columns={"price": "price_a"}
)
df_b = series[inst_b][["tstamp", "price"]].rename(
columns={"price": "price_b"}
)
merged = pd.merge(df_a, df_b, on="tstamp", how="inner").sort_values(
"tstamp"
)
# Log.info(f"{self.fname()}: analyzing {pair_name}")
stats = self._compute_stats(inst_a, inst_b, pair_name, merged)
if stats:
results[pair_name] = stats
return self._rank(results)
def _compute_stats(
self,
inst_a: ExchangeInstrument,
inst_b: ExchangeInstrument,
pair_name: str,
merged: pd.DataFrame,
) -> Optional[PairStats]:
if len(merged) < 2:
return None
px_a = merged["price_a"].astype(float)
px_b = merged["price_b"].astype(float)
std_a = float(px_a.std())
std_b = float(px_b.std())
if std_a == 0 or std_b == 0:
return None
z_a = (px_a - float(px_a.mean())) / std_a
z_b = (px_b - float(px_b.mean())) / std_b
p_eg: Optional[float]
p_adf: Optional[float]
p_j: Optional[float]
trace_stat: Optional[float]
try:
p_eg = float(coint(z_a, z_b)[1])
except Exception as exc:
Log.warning(
f"{self.fname()}: EG failed for {inst_a.details_short()}/{inst_b.details_short()}: {exc}"
)
p_eg = None
try:
spread = z_a - z_b
p_adf = float(adfuller(spread, maxlag=1, regression="c")[1])
except Exception as exc:
Log.warning(
f"{self.fname()}: ADF failed for {inst_a.details_short()}/{inst_b.details_short()}: {exc}"
)
p_adf = None
try:
data = np.column_stack([z_a, z_b])
res = coint_johansen(data, det_order=0, k_ar_diff=1)
trace_stat = float(res.lr1[0])
cv10, cv5, cv1 = res.cvt[0]
if trace_stat > cv1:
p_j = 0.01
elif trace_stat > cv5:
p_j = 0.05
elif trace_stat > cv10:
p_j = 0.10
else:
p_j = 1.0
except Exception as exc:
Log.warning(
f"{self.fname()}: Johansen failed for {inst_a.details_short()}/{inst_b.details_short()}: {exc}"
)
p_j = None
trace_stat = None
return PairStats(
pair_name_=pair_name,
instrument_a_=inst_a,
instrument_b_=inst_b,
pvalue_eg_=p_eg,
pvalue_adf_=p_adf,
pvalue_j_=p_j,
trace_stat_j_=trace_stat,
)
def _rank(self, results: Dict[str, PairStats]) -> Dict[str, PairStats]:
ranked = list(results.values())
self._assign_ranks(ranked, key=lambda r: r.pvalue_eg_, attr="rank_eg_")
self._assign_ranks(ranked, key=lambda r: r.pvalue_adf_, attr="rank_adf_")
self._assign_ranks(ranked, key=lambda r: r.pvalue_j_, attr="rank_j_")
for res in ranked:
res.composite_rank_ = res.rank_eg_ + res.rank_adf_ # + res.rank_j_
ranked.sort(key=lambda r: r.composite_rank_)
return {res.pair_name_: res for res in ranked}
@staticmethod
def _normalized_pair(
inst_a: ExchangeInstrument, inst_b: ExchangeInstrument
) -> Tuple[ExchangeInstrument, ExchangeInstrument, str]:
inst_a_id = PairAnalyzer._pair_label(inst_a.instrument_id())
inst_b_id = PairAnalyzer._pair_label(inst_b.instrument_id())
if inst_a_id <= inst_b_id:
return inst_a, inst_b, f"{inst_a_id}<->{inst_b_id}"
return inst_b, inst_a, f"{inst_b_id}<->{inst_a_id}"
@staticmethod
def _pair_label(instrument_id: str) -> str:
if instrument_id.startswith("PAIR-"):
return instrument_id[len("PAIR-") :]
return instrument_id
@staticmethod
def _assign_ranks(results: List[PairStats], key, attr: str) -> None:
values = [key(r) for r in results]
sorted_vals = sorted([v for v in values if v is not None])
for res in results:
val = key(res)
if val is None:
setattr(res, attr, len(sorted_vals) + 1)
continue
rank = 1 + sum(1 for v in sorted_vals if v < val)
setattr(res, attr, rank)
class PairSelectionEngine(NamedObject):
config_: object
instruments_: List[ExchangeInstrument]
price_field_: str
fetcher_: DataFetcher
quality_: QualityChecker
analyzer_: PairAnalyzer
interval_sec_: int
history_depth_sec_: int
data_quality_cache_: List[InstrumentQuality]
pair_results_cache_: Dict[str, PairStats]
def __init__(
self,
config: Config,
instruments: List[ExchangeInstrument],
price_field: str,
) -> None:
self.config_ = config
self.instruments_ = instruments
self.price_field_ = price_field
interval_sec = int(config.get_value("interval_sec", 0))
history_depth_sec = int(config.get_value("history_depth_hours", 0)) * SecPerHour
base_url = config.get_value("cvtt_base_url", None)
assert interval_sec > 0, "interval_sec must be > 0"
assert history_depth_sec > 0, "history_depth_sec must be > 0"
assert base_url, "cvtt_base_url must be set"
self.fetcher_ = DataFetcher(
base_url=base_url,
interval_sec=interval_sec,
history_depth_sec=history_depth_sec,
)
self.quality_ = QualityChecker(interval_sec=interval_sec)
self.analyzer_ = PairAnalyzer(
price_field=price_field, interval_sec=interval_sec
)
self.interval_sec_ = interval_sec
self.history_depth_sec_ = history_depth_sec
self.data_quality_cache_ = []
self.pair_results_cache_ = {}
async def run_once(self) -> None:
quality_results: List[InstrumentQuality] = []
price_series: Dict[ExchangeInstrument, pd.DataFrame] = {}
for inst in self.instruments_:
exch_acct = inst.user_data_.get("exch_acct") or inst.exchange_id_
aggr = self.fetcher_.fetch(exch_acct=exch_acct, inst=inst)
q = self.quality_.evaluate(inst, aggr)
quality_results.append(q)
if q.status_ != "PASS":
continue
df = self._to_dataframe(aggr, inst)
if len(df) > 0:
price_series[inst] = df
self.data_quality_cache_ = quality_results
self.pair_results_cache_ = self.analyzer_.analyze(price_series)
def _to_dataframe(
self, aggr: List[MdTradesAggregate], inst: ExchangeInstrument
) -> pd.DataFrame:
rows: List[Dict[str, Any]] = []
for item in aggr:
rows.append(
{
"tstamp": pd.to_datetime(item.aggr_time_ns_, unit="ns", utc=True),
"price": self._extract_price(item, inst),
"num_trades": item.num_trades_,
}
)
df = pd.DataFrame(rows)
return df.sort_values("tstamp").reset_index(drop=True)
def _extract_price(
self, aggr: MdTradesAggregate, inst: ExchangeInstrument
) -> float:
return _extract_price_from_fields(
price_field=self.price_field_,
inst=inst,
open=aggr.open_,
high=aggr.high_,
low=aggr.low_,
close=aggr.close_,
vwap=aggr.vwap_,
)
def sleep_seconds_until_next_cycle(self) -> float:
now_ns = current_nanoseconds()
interval_ns = self.interval_sec_ * NanoPerSec
next_boundary = (now_ns // interval_ns + 1) * interval_ns
return max(0.0, (next_boundary - now_ns) / NanoPerSec)
def quality_dicts(self) -> List[Dict[str, Any]]:
res: List[Dict[str, Any]] = []
for q in self.data_quality_cache_:
res.append(
{
"instrument": q.instrument_.instrument_id(),
"record_count": q.record_count_,
"latest_tstamp": (
q.latest_tstamp_.isoformat() if q.latest_tstamp_ else None
),
"status": q.status_,
"reason": q.reason_,
}
)
return res
def pair_dicts(self) -> Dict[str, Dict[str, Any]]:
return {
pair_name: stats.as_dict()
for pair_name, stats in self.pair_results_cache_.items()
}
class PairSelectionBacktest(NamedObject):
config_: object
instruments_: List[ExchangeInstrument]
price_field_: str
input_db_: str
output_db_: str
interval_sec_: int
history_depth_hours_: int
quality_: QualityChecker
analyzer_: PairAnalyzer
inst_by_key_: Dict[Tuple[str, str], ExchangeInstrument]
inst_by_id_: Dict[str, Optional[ExchangeInstrument]]
ambiguous_ids_: Set[str]
def __init__(
self,
config: Config,
instruments: List[ExchangeInstrument],
price_field: str,
input_db: str,
output_db: str,
) -> None:
self.config_ = config
self.instruments_ = instruments
self.price_field_ = price_field
self.input_db_ = input_db
self.output_db_ = output_db
interval_sec = int(config.get_value("interval_sec", 0))
if interval_sec <= 0:
Log.warning(
f"{self.fname()}: interval_sec not set; defaulting to 60 seconds"
)
interval_sec = 60
history_depth_hours = int(config.get_value("history_depth_hours", 0))
assert history_depth_hours > 0, "history_depth_hours must be > 0"
self.interval_sec_ = interval_sec
self.history_depth_hours_ = history_depth_hours
self.quality_ = QualityChecker(interval_sec=interval_sec)
self.analyzer_ = PairAnalyzer(
price_field=price_field, interval_sec=interval_sec
)
self.inst_by_key_ = {
(inst.exchange_id_, inst.instrument_id()): inst for inst in instruments
}
self.inst_by_id_ = {}
self.ambiguous_ids_ = set()
for inst in instruments:
inst_id = inst.instrument_id()
if inst_id in self.inst_by_id_:
existing = self.inst_by_id_[inst_id]
if existing is not None and existing.exchange_id_ != inst.exchange_id_:
self.inst_by_id_[inst_id] = None
self.ambiguous_ids_.add(inst_id)
elif inst_id not in self.ambiguous_ids_:
self.inst_by_id_[inst_id] = inst
if self.ambiguous_ids_:
Log.warning(
f"{self.fname()}: ambiguous instrument_id(s) without exchange_id: "
f"{sorted(self.ambiguous_ids_)}"
)
def run(self) -> None:
df = self._load_input_df()
if df.empty:
Log.warning(f"{self.fname()}: no rows in md_1min_bars")
return
df = self._filter_instruments(df)
if df.empty:
Log.warning(f"{self.fname()}: no rows after instrument filtering")
return
conn = self._init_output_db()
try:
self._run_backtest(df, conn)
finally:
conn.commit()
conn.close()
def _load_input_df(self) -> pd.DataFrame:
if not os.path.exists(self.input_db_):
raise FileNotFoundError(f"input_db not found: {self.input_db_}")
with sqlite3.connect(self.input_db_) as conn:
df = pd.read_sql_query(
"""
SELECT
tstamp,
tstamp_ns,
exchange_id,
instrument_id,
open,
high,
low,
close,
volume,
vwap,
num_trades
FROM md_1min_bars
""",
conn,
)
if df.empty:
return df
ts_ns = pd.to_datetime(df["tstamp_ns"], unit="ns", utc=True, errors="coerce")
ts_txt = pd.to_datetime(df["tstamp"], utc=True, errors="coerce")
df["tstamp"] = ts_ns.fillna(ts_txt)
df = df.dropna(subset=["tstamp", "instrument_id"]).copy()
df["exchange_id"] = df["exchange_id"].fillna("")
df["instrument_id"] = df["instrument_id"].astype(str)
df["tstamp_ns"] = df["tstamp"].astype("int64")
return df.sort_values("tstamp").reset_index(drop=True)
def _filter_instruments(self, df: pd.DataFrame) -> pd.DataFrame:
instrument_ids = {inst.instrument_id() for inst in self.instruments_}
df = df[df["instrument_id"].isin(instrument_ids)].copy()
if "exchange_id" in df.columns:
exchange_ids = {inst.exchange_id_ for inst in self.instruments_}
df = df[
(df["exchange_id"].isin(exchange_ids)) | (df["exchange_id"] == "")
].copy()
return df
def _init_output_db(self) -> sqlite3.Connection:
if os.path.exists(self.output_db_):
os.remove(self.output_db_)
conn = sqlite3.connect(self.output_db_)
conn.execute(
"""
CREATE TABLE pair_selection_history (
tstamp TEXT,
tstamp_ns INTEGER,
pair_name TEXT,
exchange_a TEXT,
instrument_a TEXT,
exchange_b TEXT,
instrument_b TEXT,
pvalue_eg REAL,
pvalue_adf REAL,
pvalue_j REAL,
trace_stat_j REAL,
rank_eg INTEGER,
rank_adf INTEGER,
rank_j INTEGER,
composite_rank REAL
)
"""
)
conn.execute(
"""
CREATE INDEX idx_pair_selection_history_pair_name
ON pair_selection_history (pair_name)
"""
)
conn.execute(
"""
CREATE UNIQUE INDEX idx_pair_selection_history_tstamp_pair
ON pair_selection_history (tstamp, pair_name)
"""
)
conn.commit()
return conn
def _resolve_instrument(
self, exchange_id: str, instrument_id: str
) -> Optional[ExchangeInstrument]:
if exchange_id:
inst = self.inst_by_key_.get((exchange_id, instrument_id))
if inst is not None:
return inst
inst = self.inst_by_id_.get(instrument_id)
if inst is None and instrument_id in self.ambiguous_ids_:
return None
return inst
def _build_day_series(
self, df_day: pd.DataFrame
) -> Dict[ExchangeInstrument, pd.DataFrame]:
series: Dict[ExchangeInstrument, pd.DataFrame] = {}
group_cols = ["exchange_id", "instrument_id"]
for key, group in df_day.groupby(group_cols, dropna=False):
exchange_id, instrument_id = key
inst = self._resolve_instrument(str(exchange_id or ""), str(instrument_id))
if inst is None:
continue
df_inst = group.copy()
df_inst["price"] = [
_extract_price_from_fields(
price_field=self.price_field_,
inst=inst,
open=float(row.open), #type: ignore
high=float(row.high), #type: ignore
low=float(row.low), #type: ignore
close=float(row.close), #type: ignore
vwap=float(row.vwap),#type: ignore
)
for row in df_inst.itertuples(index=False)
]
df_inst = df_inst[["tstamp", "tstamp_ns", "price", "num_trades"]]
if inst in series:
series[inst] = pd.concat([series[inst], df_inst], ignore_index=True)
else:
series[inst] = df_inst
for inst in list(series.keys()):
series[inst] = series[inst].sort_values("tstamp").reset_index(drop=True)
return series
def _run_backtest(self, df: pd.DataFrame, conn: sqlite3.Connection) -> None:
window_minutes = self.history_depth_hours_ * 60
window_td = pd.Timedelta(minutes=window_minutes)
step_td = pd.Timedelta(seconds=self.interval_sec_)
df = df.copy()
df["day"] = df["tstamp"].dt.normalize()
days = sorted(df["day"].unique())
for day in days:
day_label = pd.Timestamp(day).date()
df_day = df[df["day"] == day]
t0 = df_day["tstamp"].min()
t_last = df_day["tstamp"].max()
if t_last - t0 < window_td:
Log.warning(
f"{self.fname()}: skipping {day_label} (insufficient data)"
)
continue
day_series = self._build_day_series(df_day)
if len(day_series) < 2:
Log.warning(
f"{self.fname()}: skipping {day_label} (insufficient instruments)"
)
continue
start = t0
expected_end = start + window_td
while expected_end <= t_last:
window_slices: Dict[ExchangeInstrument, pd.DataFrame] = {}
ts: Optional[pd.Timestamp] = None
for inst, df_inst in day_series.items():
df_win = df_inst[
(df_inst["tstamp"] >= start)
& (df_inst["tstamp"] < expected_end)
]
if df_win.empty:
continue
window_slices[inst] = df_win
last_ts = df_win["tstamp"].iloc[-1]
if ts is None or last_ts > ts:
ts = last_ts
if window_slices and ts is not None:
price_series: Dict[ExchangeInstrument, pd.DataFrame] = {}
for inst, df_win in window_slices.items():
aggr = self._to_backtest_aggregates(df_win)
q = self.quality_.evaluate(
inst=inst, aggr=aggr, now_ts=ts
)
if q.status_ != "PASS":
continue
price_series[inst] = df_win[["tstamp", "price"]]
pair_results = self.analyzer_.analyze(price_series)
Log.info(f"{self.fname()}: Saving Results for window ending {ts}")
self._insert_results(conn, ts, pair_results)
start = start + step_td
expected_end = start + window_td
@staticmethod
def _to_backtest_aggregates(df_win: pd.DataFrame) -> List[BacktestAggregate]:
aggr: List[BacktestAggregate] = []
for tstamp_ns, num_trades in zip(df_win["tstamp_ns"], df_win["num_trades"]):
nt = None if pd.isna(num_trades) else int(num_trades)
aggr.append(
BacktestAggregate(aggr_time_ns_=int(tstamp_ns), num_trades_=nt)
)
return aggr
@staticmethod
def _insert_results(
conn: sqlite3.Connection,
ts: pd.Timestamp,
pair_results: Dict[str, PairStats],
) -> None:
if not pair_results:
return
iso = ts.isoformat()
ns = int(ts.value)
rows = []
for pair_name in sorted(pair_results.keys()):
stats = pair_results[pair_name]
rows.append(
(
iso,
ns,
pair_name,
stats.instrument_a_.exchange_id_,
stats.instrument_a_.instrument_id(),
stats.instrument_b_.exchange_id_,
stats.instrument_b_.instrument_id(),
stats.pvalue_eg_,
stats.pvalue_adf_,
stats.pvalue_j_,
stats.trace_stat_j_,
stats.rank_eg_,
stats.rank_adf_,
stats.rank_j_,
stats.composite_rank_,
)
)
conn.executemany(
"""
INSERT INTO pair_selection_history (
tstamp,
tstamp_ns,
pair_name,
exchange_a,
instrument_a,
exchange_b,
instrument_b,
pvalue_eg,
pvalue_adf,
pvalue_j,
trace_stat_j,
rank_eg,
rank_adf,
rank_j,
composite_rank
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
rows,
)
conn.commit()
class PairSelector(NamedObject):
instruments_: List[ExchangeInstrument]
engine_: PairSelectionEngine
rest_service_: Optional[RestService]
backtest_: Optional[PairSelectionBacktest]
def __init__(self) -> None:
App.instance().add_cmdline_arg("--oneshot", action="store_true", default=False)
App.instance().add_cmdline_arg("--backtest", action="store_true", default=False)
App.instance().add_cmdline_arg("--input_db", default=None)
App.instance().add_cmdline_arg("--output_db", default=None)
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:
cfg = CvttAppConfig.instance()
self.instruments_ = self._load_instruments(cfg)
price_field = cfg.get_value("model/stat_model_price", "close")
self.backtest_ = None
self.rest_service_ = None
if App.instance().get_argument("backtest", False):
input_db = App.instance().get_argument("input_db", None)
output_db = App.instance().get_argument("output_db", None)
if not input_db or not output_db:
raise ValueError(
"--input_db and --output_db are required when --backtest is set"
)
self.backtest_ = PairSelectionBacktest(
config=cfg,
instruments=self.instruments_,
price_field=price_field,
input_db=input_db,
output_db=output_db,
)
return
self.engine_ = PairSelectionEngine(
config=cfg,
instruments=self.instruments_,
price_field=price_field,
)
self.rest_service_ = RestService(config_key="/api/REST")
self.rest_service_.add_handler("GET", "/data_quality", self._on_data_quality)
self.rest_service_.add_handler(
"GET", "/pair_selection", self._on_pair_selection
)
def _load_instruments(self, cfg: CvttAppConfig) -> List[ExchangeInstrument]:
instruments_cfg = cfg.get_value("instruments", [])
instruments: List[ExchangeInstrument] = []
assert len(instruments_cfg) >= 2, "at least two instruments required"
for item in instruments_cfg:
if isinstance(item, str):
parts = item.split(":", 1)
if len(parts) != 2:
raise ValueError(f"invalid instrument format: {item}")
exch_acct, instrument_id = parts
elif isinstance(item, dict):
exch_acct = item.get("exch_acct", "")
instrument_id = item.get("instrument_id", "")
if not exch_acct or not instrument_id:
raise ValueError(f"invalid instrument config: {item}")
else:
raise ValueError(f"unsupported instrument entry: {item}")
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 {exch_acct}:{instrument_id}"
exch_inst.user_data_["exch_acct"] = exch_acct
instruments.append(exch_inst)
return instruments
async def run(self) -> None:
if App.instance().get_argument("backtest", False):
if self.backtest_ is None:
raise RuntimeError("backtest runner not initialized")
self.backtest_.run()
return
oneshot = App.instance().get_argument("oneshot", False)
while True:
await self.engine_.run_once()
if oneshot:
break
sleep_for = self.engine_.sleep_seconds_until_next_cycle()
await asyncio.sleep(sleep_for)
async def _on_data_quality(self, request: web.Request) -> web.Response:
fmt = request.query.get("format", "html").lower()
quality = self.engine_.quality_dicts()
if fmt == "json":
return web.json_response(quality)
return web.Response(
text=HtmlRenderer.render_data_quality(quality), content_type="text/html"
)
async def _on_pair_selection(self, request: web.Request) -> web.Response:
fmt = request.query.get("format", "html").lower()
pairs = self.engine_.pair_dicts()
if fmt == "json":
return web.json_response(pairs)
return web.Response(
text=HtmlRenderer.render_pairs(pairs), content_type="text/html"
)
if __name__ == "__main__":
App()
CvttAppConfig()
PairSelector()
App.instance().run()
-201
View File
@@ -1,201 +0,0 @@
aiohttp>=3.8.4
aiosignal>=1.3.1
async-timeout>=4.0.2
attrs>=21.2.0
beautifulsoup4>=4.10.0
black>=23.3.0
flake8>=6.0.0
certifi>=2020.6.20
chardet>=4.0.0
charset-normalizer>=3.1.0
click>=8.0.3
colorama>=0.4.4
configobj>=5.0.6
cryptography>=3.4.8
distro>=1.7.0
docker>=5.0.3
dockerpty>=0.4.1
docopt>=0.6.2
eyeD3>=0.8.10
filelock>=3.6.0
frozenlist>=1.3.3
grpcio>=1.30.2
hjson>=3.0.2
html5lib>=1.1
httplib2>=0.20.2
idna>=3.3
ipython>=8.18.1
ipywidgets>=8.1.1
ifaddr>=0.1.7
IMDbPY>=2021.4.18
ipykernel>=6.29.5
jeepney>=0.7.1
jsonschema>=3.2.0
jupyter>=1.0.0
keyring>=23.5.0
launchpadlib>=1.10.16
lazr.restfulclient>=0.14.4
lazr.uri>=1.0.6
lxml>=4.8.0
Mako>=1.1.3
Markdown>=3.3.6
MarkupSafe>=2.0.1
matplotlib>=3.10.3
more-itertools>=8.10.0
multidict>=6.0.4
mypy>=0.942
mypy-extensions>=0.4.3
nbformat>=5.10.2
netaddr>=0.8.0
######### netifaces>=0.11.0
numpy>=1.26.4,<2.3.0
oauthlib>=3.2.0
packaging>=23.1
pandas>=2.2.3
pathspec>=0.11.1
pexpect>=4.8.0
Pillow>=9.0.1
platformdirs>=3.2.0
plotly>=5.19.0
protobuf>=3.12.4
psutil>=5.9.0
ptyprocess>=0.7.0
pycurl>=7.44.1
pyelftools>=0.27
Pygments>=2.11.2
pyparsing>=2.4.7
pyrsistent>=0.18.1
python-debian>=0.1.43 #+ubuntu1.1
python-dotenv>=0.19.2
python-magic>=0.4.24
python-xlib>=0.29
pyxdg>=0.27
PyYAML>=6.0
reportlab>=3.6.8
requests>=2.25.1
requests-file>=1.5.1
scipy<1.13.0
seaborn>=0.13.2
SecretStorage>=3.3.1
setproctitle>=1.2.2
simpleeval>=1.0.3
six>=1.16.0
soupsieve>=2.3.1
ssh-import-id>=5.11
statsmodels>=0.14.4
texttable>=1.6.4
tldextract>=3.1.2
tomli>=1.2.2
######## typed-ast>=1.4.3
types-aiofiles>=0.1
types-annoy>=1.17
types-appdirs>=1.4
types-atomicwrites>=1.4
types-aws-xray-sdk>=2.8
types-babel>=2.9
types-backports-abc>=0.5
types-backports.ssl-match-hostname>=3.7
types-beautifulsoup4>=4.10
types-bleach>=4.1
types-boto>=2.49
types-braintree>=4.11
types-cachetools>=4.2
types-caldav>=0.8
types-certifi>=2020.4
types-characteristic>=14.3
types-chardet>=4.0
types-click>=7.1
types-click-spinner>=0.1
types-colorama>=0.4
types-commonmark>=0.9
types-contextvars>=0.1
types-croniter>=1.0
types-cryptography>=3.3
types-dataclasses>=0.1
types-dateparser>=1.0
types-DateTimeRange>=0.1
types-decorator>=0.1
types-Deprecated>=1.2
types-docopt>=0.6
types-docutils>=0.17
types-editdistance>=0.5
types-emoji>=1.2
types-entrypoints>=0.3
types-enum34>=1.1
types-filelock>=3.2
types-first>=2.0
types-Flask>=1.1
types-freezegun>=1.1
types-frozendict>=0.1
types-futures>=3.3
types-html5lib>=1.1
types-httplib2>=0.19
types-humanfriendly>=9.2
types-ipaddress>=1.0
types-itsdangerous>=1.1
types-JACK-Client>=0.1
types-Jinja2>=2.11
types-jmespath>=0.10
types-jsonschema>=3.2
types-Markdown>=3.3
types-MarkupSafe>=1.1
types-mock>=4.0
types-mypy-extensions>=0.4
types-mysqlclient>=2.0
types-oauthlib>=3.1
types-orjson>=3.6
types-paramiko>=2.7
types-Pillow>=8.3
types-polib>=1.1
types-prettytable>=2.1
types-protobuf>=3.17
types-psutil>=5.8
types-psycopg2>=2.9
types-pyaudio>=0.2
types-pycurl>=0.1
types-pyfarmhash>=0.2
types-Pygments>=2.9
types-PyMySQL>=1.0
types-pyOpenSSL>=20.0
types-pyRFC3339>=0.1
types-pysftp>=0.2
types-pytest-lazy-fixture>=0.6
types-python-dateutil>=2.8
types-python-gflags>=3.1
types-python-nmap>=0.6
types-python-slugify>=5.0
types-pytz>=2021.1
types-pyvmomi>=7.0
types-PyYAML>=5.4
types-redis>=3.5
types-requests>=2.25
types-retry>=0.9
types-seaborn>0.13.2
types-selenium>=3.141
types-Send2Trash>=1.8
types-setuptools>=57.4
types-simplejson>=3.17
types-singledispatch>=3.7
types-six>=1.16
types-slumber>=0.7
types-stripe>=2.59
types-tabulate>=0.8
types-termcolor>=1.1
types-toml>=0.10
types-toposort>=1.6
types-ttkthemes>=3.2
types-typed-ast>=1.4
types-tzlocal>=0.1
types-ujson>=0.1
types-vobject>=0.9
types-waitress>=0.1
types-Werkzeug>=1.0
types-xxhash>=2.0
typing-extensions>=3.10.0.2
Unidecode>=1.3.3
urllib3>=1.26.5
wadllib>=1.3.6
webencodings>=0.5.1
websocket-client>=1.2.3
yarl>=1.9.1
zipp>=1.0.0
@@ -1,311 +0,0 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Pair Selection History\n",
"\n",
"Interactive notebook for exploring pair selection history from a SQLite database.\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"**Usage**\n",
"- Enter the SQLite `db_path` (file path).\n",
"- Click `Load pairs` to populate the dropdown.\n",
"- Select a `pair_name`, then click `Plot`.\n"
]
},
{
"cell_type": "markdown",
"id": "668ebf19",
"metadata": {},
"source": [
"# Settings"
]
},
{
"cell_type": "code",
"execution_count": 1,
"id": "c78db847",
"metadata": {},
"outputs": [],
"source": [
"import sqlite3\n",
"from pathlib import Path\n",
"\n",
"import pandas as pd\n",
"import plotly.express as px\n",
"import ipywidgets as widgets\n",
"from IPython.display import display\n"
]
},
{
"cell_type": "markdown",
"id": "e7ac6adc",
"metadata": {},
"source": [
"# Data Loading"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "766bcf9f",
"metadata": {},
"outputs": [
{
"data": {
"application/vnd.jupyter.widget-view+json": {
"model_id": "e0b30b1abd1b440b832fdaaa6cce8f76",
"version_major": 2,
"version_minor": 0
},
"text/plain": [
"VBox(children=(Text(value='', description='pair_db', layout=Layout(width='80%'), placeholder='/path/to/pairs.d…"
]
},
"metadata": {},
"output_type": "display_data"
},
{
"data": {
"application/vnd.jupyter.widget-view+json": {
"model_id": "15679f9015854d5fa7119210094fbbc8",
"version_major": 2,
"version_minor": 0
},
"text/plain": [
"Output()"
]
},
"metadata": {},
"output_type": "display_data"
}
],
"source": [
"db_path = widgets.Text(\n",
" value='',\n",
" placeholder='/path/to/pairs.db',\n",
" description='pair_db',\n",
" layout=widgets.Layout(width='80%')\n",
")\n",
"\n",
"md_db_path = widgets.Text(\n",
" value='',\n",
" placeholder='/path/to/market_data.db',\n",
" description='md_db',\n",
" layout=widgets.Layout(width='80%')\n",
")\n",
"\n",
"load_button = widgets.Button(description='Load pairs', button_style='info')\n",
"plot_button = widgets.Button(description='Plot', button_style='primary')\n",
"\n",
"pair_name = widgets.Dropdown(\n",
" options=[],\n",
" value=None,\n",
" description='pair_name',\n",
" layout=widgets.Layout(width='80%')\n",
")\n",
"\n",
"status = widgets.HTML(value='')\n",
"output = widgets.Output()\n",
"\n",
"controls = widgets.VBox([\n",
" db_path,\n",
" md_db_path,\n",
" widgets.HBox([load_button, plot_button]),\n",
" pair_name,\n",
" status,\n",
"])\n",
"\n",
"display(controls, output)\n"
]
},
{
"cell_type": "markdown",
"id": "a4d47855",
"metadata": {},
"source": [
"# Processing"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "2c710f51",
"metadata": {},
"outputs": [],
"source": [
"PLOT_WIDTH = 1100\n",
"PLOT_HEIGHT = 320\n",
"\n",
"def _connect(path: str):\n",
" if not path:\n",
" raise ValueError('Please provide db_path.')\n",
" p = Path(path).expanduser().resolve()\n",
" if not p.exists():\n",
" raise FileNotFoundError(f'Database not found: {p}')\n",
" return sqlite3.connect(p)\n",
"\n",
"\n",
"def _parse_tstamp(series: pd.Series) -> pd.Series:\n",
" return pd.to_datetime(series, utc=True, errors='coerce').dt.tz_convert(None)\n",
"\n",
"\n",
"def _style_fig(fig, tmin, tmax):\n",
" fig.update_layout(\n",
" legend=dict(orientation='h', yanchor='bottom', y=1.02, xanchor='left', x=0),\n",
" margin=dict(l=50, r=20, t=60, b=40),\n",
" height=PLOT_HEIGHT,\n",
" width=PLOT_WIDTH,\n",
" )\n",
" fig.update_xaxes(range=[tmin, tmax])\n",
"\n",
"\n",
"def _load_pairs(_=None):\n",
" status.value = ''\n",
" with output:\n",
" output.clear_output()\n",
" try:\n",
" with _connect(db_path.value) as conn:\n",
" rows = conn.execute(\n",
" \"SELECT pair_name \"\n",
" \"FROM pair_selection_history \"\n",
" \"GROUP BY pair_name \"\n",
" \"ORDER BY SUM(composite_rank), pair_name\"\n",
" ).fetchall()\n",
" options = [r[0] for r in rows]\n",
" pair_name.options = options\n",
" pair_name.value = options[0] if options else None\n",
" status.value = f'Loaded {len(options)} pairs.'\n",
" except Exception as exc:\n",
" status.value = f\"<span style='color:#b00'>Error: {exc}</span>\"\n",
"\n",
"\n",
"def _plot(_=None):\n",
" status.value = ''\n",
" with output:\n",
" output.clear_output()\n",
" try:\n",
" if not pair_name.value:\n",
" raise ValueError('Please select a pair_name.')\n",
" if not md_db_path.value:\n",
" raise ValueError('Please provide md_db path.')\n",
" query = (\n",
" 'SELECT tstamp, pvalue_eg, pvalue_adf, rank_eg, rank_adf, '\n",
" 'exchange_a, instrument_a, exchange_b, instrument_b '\n",
" 'FROM pair_selection_history '\n",
" 'WHERE pair_name = ? '\n",
" 'ORDER BY tstamp'\n",
" )\n",
" with _connect(db_path.value) as conn:\n",
" df = pd.read_sql_query(query, conn, params=(pair_name.value,))\n",
" if df.empty:\n",
" raise ValueError('No data for selected pair_name.')\n",
" df['tstamp'] = _parse_tstamp(df['tstamp'])\n",
" df = df.dropna(subset=['tstamp'])\n",
" if df.empty:\n",
" raise ValueError('No valid timestamps in pair selection data.')\n",
" tmin = df['tstamp'].min()\n",
" tmax = df['tstamp'].max()\n",
"\n",
" first_row = df.dropna(subset=['exchange_a', 'instrument_a', 'exchange_b', 'instrument_b']).iloc[0]\n",
" ex_a = first_row['exchange_a']\n",
" id_a = first_row['instrument_a']\n",
" ex_b = first_row['exchange_b']\n",
" id_b = first_row['instrument_b']\n",
"\n",
" fig_p = px.line(\n",
" df,\n",
" x='tstamp',\n",
" y=['pvalue_eg', 'pvalue_adf'],\n",
" title=f'P-Values Over Time: {pair_name.value}',\n",
" labels={'value': 'p-value', 'variable': 'metric', 'tstamp': 'timestamp'}\n",
" )\n",
" fig_p.update_layout(legend_title_text='metric')\n",
" _style_fig(fig_p, tmin, tmax)\n",
"\n",
" md_query = (\n",
" 'SELECT tstamp, close FROM md_1min_bars '\n",
" 'WHERE exchange_id = ? AND instrument_id = ? '\n",
" 'ORDER BY tstamp'\n",
" )\n",
" with _connect(md_db_path.value) as md_conn:\n",
" md_a = pd.read_sql_query(md_query, md_conn, params=(ex_a, id_a))\n",
" md_b = pd.read_sql_query(md_query, md_conn, params=(ex_b, id_b))\n",
" if md_a.empty or md_b.empty:\n",
" raise ValueError('Market data not found for selected instruments.')\n",
" md_a['tstamp'] = _parse_tstamp(md_a['tstamp'])\n",
" md_b['tstamp'] = _parse_tstamp(md_b['tstamp'])\n",
" md_a = md_a.dropna(subset=['tstamp', 'close'])\n",
" md_b = md_b.dropna(subset=['tstamp', 'close'])\n",
" md_a = md_a[(md_a['tstamp'] >= tmin) & (md_a['tstamp'] <= tmax)]\n",
" md_b = md_b[(md_b['tstamp'] >= tmin) & (md_b['tstamp'] <= tmax)]\n",
" if md_a.empty or md_b.empty:\n",
" raise ValueError('Market data is outside the pair selection time range.')\n",
" md_a = md_a.sort_values('tstamp')\n",
" md_b = md_b.sort_values('tstamp')\n",
" md_a['scaled_close'] = (md_a['close'] - md_a['close'].iloc[0]) / md_a['close'].iloc[0] * 100\n",
" md_b['scaled_close'] = (md_b['close'] - md_b['close'].iloc[0]) / md_b['close'].iloc[0] * 100\n",
"\n",
" md_plot = pd.DataFrame({\n",
" 'tstamp': md_a['tstamp'],\n",
" f'{ex_a}:{id_a}': md_a['scaled_close'],\n",
" })\n",
" md_plot = md_plot.merge(\n",
" pd.DataFrame({\n",
" 'tstamp': md_b['tstamp'],\n",
" f'{ex_b}:{id_b}': md_b['scaled_close'],\n",
" }),\n",
" on='tstamp',\n",
" how='outer'\n",
" ).sort_values('tstamp')\n",
"\n",
" fig_m = px.line(\n",
" md_plot,\n",
" x='tstamp',\n",
" y=[f'{ex_a}:{id_a}', f'{ex_b}:{id_b}'],\n",
" title='Scaled Close Price Change (%)',\n",
" labels={'value': 'scaled % change', 'variable': 'instrument', 'tstamp': 'timestamp'}\n",
" )\n",
" fig_m.update_layout(legend_title_text='instrument')\n",
" _style_fig(fig_m, tmin, tmax)\n",
"\n",
" with output:\n",
" display(fig_p)\n",
" display(fig_m)\n",
" except Exception as exc:\n",
" status.value = f\"<span style='color:#b00'>Error: {exc}</span>\"\n",
"\n",
"\n",
"load_button.on_click(_load_pairs)\n",
"plot_button.on_click(_plot)\n"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "python3.12-venv",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.9"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
@@ -1,19 +1,98 @@
from __future__ import annotations
import asyncio
from typing import Any, Dict, List
from aiohttp import web
from cvttpy_tools.base.app import App
from cvttpy_tools.base.base import NamedObject
from cvttpy_tools.base.config import CvttAppConfig
from cvttpy_tools.app import App
from cvttpy_tools.base import NamedObject
from cvttpy_tools.config import CvttAppConfig
from cvttpy_tools.logger import Log
from cvttpy_tools.web.rest_service import RestService
from cvttpy_trading.trading.exchange_config import ExchangeAccounts
from cvttpy_trading.trading.instrument import ExchangeInstrument
from pairs_trading.lib.pair_selector_engine import PairSelectionEngine
class HtmlRenderer(NamedObject):
class PairSelector(NamedObject):
instruments_: List[ExchangeInstrument]
engine_: PairSelectionEngine
rest_service_: RestService
def __init__(self) -> None:
pass
App.instance().add_cmdline_arg("--oneshot", action="store_true", default=False)
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:
cfg = CvttAppConfig.instance()
self.instruments_ = self._load_instruments(cfg)
price_field = cfg.get_value("model/stat_model_price", "close")
self.engine_ = PairSelectionEngine(
config=cfg,
instruments=self.instruments_,
price_field=price_field,
)
self.rest_service_ = RestService(config_key="/api/REST")
self.rest_service_.add_handler("GET", "/data_quality", self._on_data_quality)
self.rest_service_.add_handler("GET", "/pair_selection", self._on_pair_selection)
def _load_instruments(self, cfg: CvttAppConfig) -> List[ExchangeInstrument]:
instruments_cfg = cfg.get_value("instruments", [])
instruments: List[ExchangeInstrument] = []
assert len(instruments_cfg) >= 2, "at least two instruments required"
for item in instruments_cfg:
if isinstance(item, str):
parts = item.split(":", 1)
if len(parts) != 2:
raise ValueError(f"invalid instrument format: {item}")
exch_acct, instrument_id = parts
elif isinstance(item, dict):
exch_acct = item.get("exch_acct", "")
instrument_id = item.get("instrument_id", "")
if not exch_acct or not instrument_id:
raise ValueError(f"invalid instrument config: {item}")
else:
raise ValueError(f"unsupported instrument entry: {item}")
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 {exch_acct}:{instrument_id}"
exch_inst.user_data_["exch_acct"] = exch_acct
instruments.append(exch_inst)
return instruments
async def run(self) -> None:
oneshot = App.instance().get_argument("oneshot", False)
while True:
await self.engine_.run_once()
if oneshot:
break
sleep_for = self.engine_.sleep_seconds_until_next_cycle()
await asyncio.sleep(sleep_for)
async def _on_data_quality(self, request: web.Request) -> web.Response:
fmt = request.query.get("format", "html").lower()
quality = self.engine_.quality_dicts()
if fmt == "json":
return web.json_response(quality)
return web.Response(text=HtmlRenderer.render_data_quality_html(quality), content_type="text/html")
async def _on_pair_selection(self, request: web.Request) -> web.Response:
fmt = request.query.get("format", "html").lower()
pairs = self.engine_.pair_dicts()
if fmt == "json":
return web.json_response(pairs)
return web.Response(text=HtmlRenderer.render_pair_selection_html(pairs), content_type="text/html")
class HtmlRenderer:
@staticmethod
def render_data_quality(quality: List[Dict[str, Any]]) -> str:
def render_data_quality_html(quality: List[Dict[str, Any]]) -> str:
rows = "".join(
f"<tr>"
f"<td>{q.get('instrument','')}</td>"
@@ -50,15 +129,16 @@ class HtmlRenderer(NamedObject):
"""
@staticmethod
def render_pairs(pairs: Dict[str, Dict[str, Any]]) -> str:
def render_pair_selection_html(pairs: List[Dict[str, Any]]) -> str:
if not pairs:
body = "<p>No pairs available. Check data quality and try again.</p>"
else:
body_rows = []
for pair_name, p in pairs.items():
for p in pairs:
body_rows.append(
"<tr>"
f"<td>{pair_name}</td>"
f"<td>{p.get('instrument_a','')}</td>"
f"<td>{p.get('instrument_b','')}</td>"
f"<td data-value='{p.get('rank_eg',0)}'>{p.get('rank_eg','')}</td>"
f"<td data-value='{p.get('rank_adf',0)}'>{p.get('rank_adf','')}</td>"
f"<td data-value='{p.get('rank_j',0)}'>{p.get('rank_j','')}</td>"
@@ -87,7 +167,8 @@ class HtmlRenderer(NamedObject):
<table id="pairs-table">
<thead>
<tr>
<th>Pair</th>
<th>Instrument A</th>
<th>Instrument B</th>
<th class="sortable" data-type="num">Rank-EG</th>
<th class="sortable" data-type="num">Rank-ADF</th>
<th class="sortable" data-type="num">Rank-J</th>
@@ -136,3 +217,10 @@ class HtmlRenderer(NamedObject):
</body>
</html>
"""
if __name__ == "__main__":
App()
CvttAppConfig()
PairSelector()
App.instance().run()
@@ -4,13 +4,13 @@ import asyncio
from typing import Callable, Coroutine, Dict, List
import aiohttp.web as web
from cvttpy_tools.base.app import App
from cvttpy_tools.base.config import Config
from cvttpy_tools.base.base import NamedObject
from cvttpy_tools.base.config import CvttAppConfig
from cvttpy_tools.base.logger import Log
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.comm.web.rest_service import RestService
from cvttpy_tools.web.rest_service import RestService
# ---
from cvttpy_trading.trading.instrument import ExchangeInstrument
-3
View File
@@ -136,9 +136,6 @@ echo "Release version: ${release_version}"
confirm
version_tag="v${release_version}"
if [ "$(git tag -l "${version_tag}")" != "" ]; then
version_tag="${version_tag}.$(date +%Y%m%d_%H%M)"
fi
version_comment="'${version_tag} ${project} ${branch} $(date +%Y-%m-%d)\n${whats_new}'"
cmd_arr=()
+43
View File
@@ -0,0 +1,43 @@
{
"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,
# ====== Trading Parameters ======
"stat_model_price": "close",
"execution_price": {
"column": "vwap",
"shift": 1,
},
"dis-equilibrium_open_trshld": 2.0,
"dis-equilibrium_close_trshld": 0.5,
"training_size": 120,
"model_class": "pairs_trading.lib.pt_strategy.models.OLSModel",
"model_data_policy_class": "pairs_trading.lib.pt_strategy.model_data_policy.ExpandingWindowDataPolicy",
# ====== 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",
}
}
+47
View File
@@ -0,0 +1,47 @@
{
"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,
# ====== Trading Parameters ======
"stat_model_price": "close",
"execution_price": {
"column": "vwap",
"shift": 1,
},
"dis-equilibrium_open_trshld": 1.75,
"dis-equilibrium_close_trshld": 0.9,
"model_class": "pairs_trading.lib.pt_strategy.models.OLSModel",
# "model_data_policy_class": "pairs_trading.lib.pt_strategy.model_data_policy.EGOptimizedWndDataPolicy",
# "model_data_policy_class": "pairs_trading.lib.pt_strategy.model_data_policy.ADFOptimizedWndDataPolicy",
"model_data_policy_class": "pairs_trading.lib.pt_strategy.model_data_policy.JohansenOptdWndDataPolicy",
"min_training_size": 60,
"max_training_size": 150,
# ====== 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",
}
}
+47
View File
@@ -0,0 +1,47 @@
{
"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,
# ====== Trading Parameters ======
"stat_model_price": "close",
"execution_price": {
"column": "vwap",
"shift": 1,
},
"dis-equilibrium_open_trshld": 1.75,
"dis-equilibrium_close_trshld": 0.9,
"model_class": "pairs_trading.lib.pt_strategy.models.OLSModel",
"training_size": 120,
"model_data_policy_class": "pairs_trading.lib.pt_strategy.model_data_policy.RollingWindowDataPolicy",
# "model_data_policy_class": "pairs_trading.lib.pt_strategy.model_data_policy.OptimizedWindowDataPolicy",
# "min_training_size": 60,
# "max_training_size": 150,
# ====== 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",
}
}
+48
View File
@@ -0,0 +1,48 @@
{
"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,
# ====== Trading Parameters ======
"stat_model_price": "close", # "vwap"
"execution_price": {
"column": "vwap",
"shift": 1,
},
"dis-equilibrium_open_trshld": 1.75,
"dis-equilibrium_close_trshld": 1.0,
"model_class": "pairs_trading.lib.pt_strategy.models.VECMModel",
"training_size": 120,
"model_data_policy_class": "pairs_trading.lib.pt_strategy.model_data_policy.RollingWindowDataPolicy",
# "model_data_policy_class": "pairs_trading.lib.pt_strategy.model_data_policy.OptimizedWindowDataPolicy",
# "min_training_size": 60,
# "max_training_size": 150,
# ====== 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",
}
}
@@ -5,19 +5,19 @@ from typing import Dict, Any, List, Optional, Set
import requests
from cvttpy_tools.base.base import NamedObject
from cvttpy_tools.base.logger import Log
from cvttpy_tools.base.config import Config
from cvttpy_tools.base.timer import Timer
from cvttpy_tools.base.timeutils import NanosT, current_seconds
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 NanosT, current_seconds
from cvttpy_tools.settings.cvtt_types import InstrumentIdT, IntervalSecT
from cvttpy_tools.web.rest_client import RESTSender
# ---
from cvttpy_trading.trading.instrument import ExchangeInstrument
from cvttpy_trading.trading.accounting.exch_account import ExchangeAccountNameT
from cvttpy_trading.trading.mkt_data.md_summary import MdTradesAggregate, MdSummary, MdSummaryCallbackT
from cvttpy_trading.trading.exchange_config import ExchangeAccounts
# ---
from pairs_trading.lib.live.rest import RESTSender
# class MdSummary(HistMdBar):
@@ -1,14 +1,14 @@
```python
from __future__ import annotations
from typing import Dict, Optional
from typing import Dict
import time
import requests
from cvttpy_tools.base.base import NamedObject
from cvttpy_tools.base import NamedObject
class RESTSender(NamedObject):
# Synchronous request sernder
session_: requests.Session
base_url_: str
@@ -26,35 +26,35 @@ class RESTSender(NamedObject):
except requests.exceptions.RequestException:
return False
def send_post(
self, endpoint: str, post_body: Dict, headers: Optional[Dict[str, str]] = None
) -> requests.Response:
def send_post(self, endpoint: str, post_body: Dict) -> requests.Response:
while not self.is_ready():
print("Waiting for FrontGateway to start...")
time.sleep(5)
if not headers:
headers = {"Content-Type": "application/json"}
url = f"{self.base_url_}/{endpoint}"
try:
return self.session_.request(
method="POST",
url=url,
json=post_body,
headers=headers,
headers={"Content-Type": "application/json"},
)
except requests.exceptions.RequestException as excpt:
raise ConnectionError(
f"Failed to send status={excpt.response.status_code} {excpt.response.text}" # type: ignore
) from excpt
def send_get(
self, endpoint: str, headers: Optional[Dict[str, str]] = None
) -> requests.Response:
if not headers:
headers = {}
def send_get(self, endpoint: str) -> requests.Response:
while not self.is_ready():
print("Waiting for FrontGateway to start...")
time.sleep(5)
url = f"{self.base_url_}/{endpoint}"
try:
return self.session_.request(method="GET", url=url, headers=headers)
return self.session_.request(method="GET", url=url)
except requests.exceptions.RequestException as excpt:
raise ConnectionError(
f"Failed to send status={excpt.response.status_code} {excpt.response.text}" # type: ignore
) from excpt
```
@@ -3,14 +3,14 @@ from enum import Enum
import requests
# import aiohttp
from cvttpy_tools.base.base import NamedObject
from cvttpy_tools.base.config import Config
from cvttpy_tools.base.logger import Log
from cvttpy_tools.base import NamedObject
from cvttpy_tools.config import Config
from cvttpy_tools.logger import Log
from cvttpy_tools.web.rest_client import RESTSender
# ---
from cvttpy_trading.trading.trading_instructions import TradingInstructions
# ---
from pairs_trading.apps.pair_trader import PairTrader
from pairs_trading.lib.live.rest import RESTSender
class TradingInstructionsSender(NamedObject):
+420
View File
@@ -0,0 +1,420 @@
from __future__ import annotations
from dataclasses import dataclass
from readline import insert_text
from typing import Any, Dict, List, Optional, Tuple
import numpy as np
import pandas as pd
from statsmodels.tsa.stattools import adfuller, coint
from statsmodels.tsa.vector_ar.vecm import coint_johansen
# ---
from cvttpy_tools.base import NamedObject
from cvttpy_tools.config import Config
from cvttpy_tools.logger import Log
from cvttpy_tools.timeutils import NanoPerSec, SecPerHour, current_nanoseconds
from cvttpy_tools.web.rest_client import RESTSender
# ---
from cvttpy_trading.trading.instrument import ExchangeInstrument
from cvttpy_trading.trading.mkt_data.md_summary import MdTradesAggregate, MdSummary
@dataclass
class InstrumentQuality(NamedObject):
instrument_: ExchangeInstrument
record_count_: int
latest_tstamp_: Optional[pd.Timestamp]
status_: str
reason_: str
@dataclass
class PairStats(NamedObject):
tstamp_: pd.Timestamp
instrument_a_: ExchangeInstrument
instrument_b_: ExchangeInstrument
pvalue_eg_: Optional[float]
pvalue_adf_: Optional[float]
pvalue_j_: Optional[float]
trace_stat_j_: Optional[float]
rank_eg_: int = 0
rank_adf_: int = 0
rank_j_: int = 0
composite_rank_: int = 0
def as_dict(self) -> Dict[str, Any]:
return {
"instrument_a": self.instrument_a_.instrument_id(),
"instrument_b": self.instrument_b_.instrument_id(),
"pvalue_eg": self.pvalue_eg_,
"pvalue_adf": self.pvalue_adf_,
"pvalue_j": self.pvalue_j_,
"trace_stat_j": self.trace_stat_j_,
"rank_eg": self.rank_eg_,
"rank_adf": self.rank_adf_,
"rank_j": self.rank_j_,
"composite_rank": self.composite_rank_,
}
class DataFetcher(NamedObject):
sender_: RESTSender
interval_sec_: int
history_depth_sec_: int
def __init__(
self,
base_url: str,
interval_sec: int,
history_depth_sec: int,
) -> None:
self.sender_ = RESTSender(base_url=base_url)
self.interval_sec_ = interval_sec
self.history_depth_sec_ = history_depth_sec
def fetch(self, exch_acct: str, inst: ExchangeInstrument) -> List[MdTradesAggregate]:
rqst_data = {
"exch_acct": exch_acct,
"instrument_id": inst.instrument_id(),
"interval_sec": self.interval_sec_,
"history_depth_sec": self.history_depth_sec_,
}
response = self.sender_.send_post(endpoint="md_summary", post_body=rqst_data)
if response.status_code not in (200, 201):
Log.error(
f"{self.fname()}: error {response.status_code} for {inst.details_short()}: {response.text}")
return []
mdsums: List[MdSummary] = MdSummary.from_REST_response(response=response)
return [
mdsum.create_md_trades_aggregate(
exch_acct=exch_acct, exch_inst=inst, interval_sec=self.interval_sec_
)
for mdsum in mdsums
]
class QualityChecker(NamedObject):
interval_sec_: int
def __init__(self, interval_sec: int) -> None:
self.interval_sec_ = interval_sec
def evaluate(self, inst: ExchangeInstrument, aggr: List[MdTradesAggregate]) -> InstrumentQuality:
if len(aggr) == 0:
return InstrumentQuality(
instrument_=inst,
record_count_=0,
latest_tstamp_=None,
status_="FAIL",
reason_="no records",
)
aggr_sorted = sorted(aggr, key=lambda a: a.aggr_time_ns_)
latest_ts = pd.to_datetime(aggr_sorted[-1].aggr_time_ns_, unit="ns", utc=True)
now_ts = pd.Timestamp.utcnow()
recency_cutoff = now_ts - pd.Timedelta(seconds=10 * self.interval_sec_)
if latest_ts <= recency_cutoff:
return InstrumentQuality(
instrument_=inst,
record_count_=len(aggr_sorted),
latest_tstamp_=latest_ts,
status_="FAIL",
reason_=f"stale: latest {latest_ts} <= cutoff {recency_cutoff}",
)
gaps_ok, reason = self._check_gaps(aggr_sorted)
status = "PASS" if gaps_ok else "FAIL"
return InstrumentQuality(
instrument_=inst,
record_count_=len(aggr_sorted),
latest_tstamp_=latest_ts,
status_=status,
reason_=reason,
)
def _check_gaps(self, aggr: List[MdTradesAggregate]) -> Tuple[bool, str]:
NUM_TRADES_THRESHOLD = 50
if len(aggr) < 2:
return True, "ok"
interval_ns = self.interval_sec_ * NanoPerSec
for idx in range(1, len(aggr)):
prev = aggr[idx - 1]
curr = aggr[idx]
delta = curr.aggr_time_ns_ - prev.aggr_time_ns_
missing_intervals = int(delta // interval_ns) - 1
if missing_intervals <= 0:
continue
prev_nt = prev.num_trades_
next_nt = curr.num_trades_
estimate = self._approximate_num_trades(prev_nt, next_nt)
if estimate > NUM_TRADES_THRESHOLD:
return False, (
f"gap of {missing_intervals} interval(s), est num_trades={estimate} > {NUM_TRADES_THRESHOLD}"
)
return True, "ok"
@staticmethod
def _approximate_num_trades(prev_nt: int, next_nt: int) -> float:
if prev_nt is None and next_nt is None:
return 0.0
if prev_nt is None:
return float(next_nt)
if next_nt is None:
return float(prev_nt)
return (prev_nt + next_nt) / 2.0
class PairOfInstruments(NamedObject):
intrument_a_: ExchangeInstrument
instrument_b_: ExchangeInstrument
stats_history_: List[PairStats]
def __init__(self, instrument_a: ExchangeInstrument, instrument_b: ExchangeInstrument) -> None:
self.instrument_a_ = instrument_a
self.instrument_b_ = instrument_b
self.stats_history_ = []
def add_stats(self, stats: PairStats) -> None:
self.stats_history_.append(stats)
class PairAnalyzer(NamedObject):
price_field_: str
interval_sec_: int
def __init__(self, price_field: str, interval_sec: int) -> None:
self.price_field_ = price_field
self.interval_sec_ = interval_sec
def analyze(self, pairs: List[PairOfInstruments], series: Dict[ExchangeInstrument, pd.DataFrame]) -> None:
for pair in pairs:
inst_a = pair.instrument_a_
inst_b = pair.instrument_b_
df_a = series[inst_a][["tstamp", "price"]].rename(
columns={"price": "price_a"}
)
df_b = series[inst_b][["tstamp", "price"]].rename(
columns={"price": "price_b"}
)
merged = pd.merge(df_a, df_b, on="tstamp", how="inner").sort_values(
"tstamp"
)
stats = self._compute_stats(inst_a, inst_b, merged)
if stats:
pair.add_stats(stats)
self._rank(pairs)
def _compute_stats(
self,
inst_a: ExchangeInstrument,
inst_b: ExchangeInstrument,
merged: pd.DataFrame,
) -> Optional[PairStats]:
if len(merged) < 2:
return None
px_a = merged["price_a"].astype(float)
px_b = merged["price_b"].astype(float)
std_a = float(px_a.std())
std_b = float(px_b.std())
if std_a == 0 or std_b == 0:
return None
z_a = (px_a - float(px_a.mean())) / std_a
z_b = (px_b - float(px_b.mean())) / std_b
p_eg: Optional[float]
p_adf: Optional[float]
p_j: Optional[float]
trace_stat: Optional[float]
try:
p_eg = float(coint(z_a, z_b)[1])
except Exception as exc:
Log.warning(f"{self.fname()}: EG failed for {inst_a.details_short()}/{inst_b.details_short()}: {exc}")
p_eg = None
try:
spread = z_a - z_b
p_adf = float(adfuller(spread, maxlag=1, regression="c")[1])
except Exception as exc:
Log.warning(f"{self.fname()}: ADF failed for {inst_a.details_short()}/{inst_b.details_short()}: {exc}")
p_adf = None
try:
data = np.column_stack([z_a, z_b])
res = coint_johansen(data, det_order=0, k_ar_diff=1)
trace_stat = float(res.lr1[0])
cv10, cv5, cv1 = res.cvt[0]
if trace_stat > cv1:
p_j = 0.01
elif trace_stat > cv5:
p_j = 0.05
elif trace_stat > cv10:
p_j = 0.10
else:
p_j = 1.0
except Exception as exc:
Log.warning(f"{self.fname()}: Johansen failed for {inst_a.details_short()}/{inst_b.details_short()}: {exc}")
p_j = None
trace_stat = None
return PairStats(
tstamp_=pd.Timestamp.utcnow(),
instrument_a_=inst_a,
instrument_b_=inst_b,
pvalue_eg_=p_eg,
pvalue_adf_=p_adf,
pvalue_j_=p_j,
trace_stat_j_=trace_stat,
)
def _rank(self, pairs: List[PairOfInstruments]) -> None:
last_stats: List[PairStats] = [pair.stats_history_[-1] for pair in pairs]
self._assign_ranks(last_stats, key=lambda r: r.pvalue_eg_, attr="rank_eg_")
self._assign_ranks(last_stats, key=lambda r: r.pvalue_adf_, attr="rank_adf_")
self._assign_ranks(last_stats, key=lambda r: r.pvalue_j_, attr="rank_j_")
for res in last_stats:
res.composite_rank_ = res.rank_eg_ + res.rank_adf_ + res.rank_j_
last_stats.sort(key=lambda r: r.composite_rank_)
@staticmethod
def _assign_ranks(
results: List[PairStats], key, attr: str
) -> None:
values = [key(r) for r in results]
sorted_vals = sorted([v for v in values if v is not None])
for res in results:
val = key(res)
if val is None:
setattr(res, attr, len(sorted_vals) + 1)
continue
rank = 1 + sum(1 for v in sorted_vals if v < val)
setattr(res, attr, rank)
class PairSelectionEngine(NamedObject):
config_: object
instruments_: List[ExchangeInstrument]
price_field_: str
fetcher_: DataFetcher
quality_: QualityChecker
analyzer_: PairAnalyzer
interval_sec_: int
history_depth_sec_: int
data_quality_cache_: List[InstrumentQuality]
pair_results_cache_: List[PairStats]
pairs_: List[PairOfInstruments]
def __init__(
self,
config: Config,
instruments: List[ExchangeInstrument],
price_field: str,
) -> None:
self.config_ = config
self.instruments_ = instruments
self.price_field_ = price_field
interval_sec = int(config.get_value("interval_sec", 0))
history_depth_sec = int(config.get_value("history_depth_hours", 0)) * SecPerHour
base_url = config.get_value("cvtt_base_url", None)
assert interval_sec > 0, "interval_sec must be > 0"
assert history_depth_sec > 0, "history_depth_sec must be > 0"
assert base_url, "cvtt_base_url must be set"
self.fetcher_ = DataFetcher(
base_url=base_url,
interval_sec=interval_sec,
history_depth_sec=history_depth_sec,
)
self.quality_ = QualityChecker(interval_sec=interval_sec)
self.analyzer_ = PairAnalyzer(price_field=price_field, interval_sec=interval_sec)
self.interval_sec_ = interval_sec
self.history_depth_sec_ = history_depth_sec
self.data_quality_cache_ = []
self.pair_results_cache_ = []
self.create_pairs(instruments)
def create_pairs(self, instruments: List[ExchangeInstrument]) -> None:
self.pairs_ = []
for i in range(len(instruments)):
for j in range(i + 1, len(instruments)):
self.pairs_.append(PairOfInstruments(instruments[i], instruments[j]))
async def run_once(self) -> None:
quality_results: List[InstrumentQuality] = []
price_series: Dict[ExchangeInstrument, pd.DataFrame] = {}
for inst in self.instruments_:
exch_acct = inst.user_data_.get("exch_acct") or inst.exchange_id_
aggr = self.fetcher_.fetch(exch_acct=exch_acct, inst=inst)
q = self.quality_.evaluate(inst, aggr)
quality_results.append(q)
if q.status_ != "PASS":
continue
df = self._to_dataframe(aggr, inst)
if len(df) > 0:
price_series[inst] = df
self.analyzer_.analyze(pairs=self.pairs_, series=price_series)
self.data_quality_cache_ = quality_results
self.pair_results_cache_ = [pair.stats_history_[-1] for pair in self.pairs_]
def _to_dataframe(self, aggr: List[MdTradesAggregate], inst: ExchangeInstrument) -> pd.DataFrame:
rows: List[Dict[str, Any]] = []
for item in aggr:
rows.append(
{
"tstamp": pd.to_datetime(item.aggr_time_ns_, unit="ns", utc=True),
"price": self._extract_price(item, inst),
"num_trades": item.num_trades_,
}
)
df = pd.DataFrame(rows)
return df.sort_values("tstamp").reset_index(drop=True)
def _extract_price(self, aggr: MdTradesAggregate, inst: ExchangeInstrument) -> float:
price_field = self.price_field_
# MdTradesAggregate inherits hist bar with fields open_, high_, low_, close_, vwap_
field_map = {
"open": aggr.open_,
"high": aggr.high_,
"low": aggr.low_,
"close": aggr.close_,
"vwap": aggr.vwap_,
}
raw = field_map.get(price_field, aggr.close_)
return inst.get_price(raw)
def sleep_seconds_until_next_cycle(self) -> float:
now_ns = current_nanoseconds()
interval_ns = self.interval_sec_ * NanoPerSec
next_boundary = (now_ns // interval_ns + 1) * interval_ns
return max(0.0, (next_boundary - now_ns) / NanoPerSec)
def quality_dicts(self) -> List[Dict[str, Any]]:
res: List[Dict[str, Any]] = []
for q in self.data_quality_cache_:
res.append(
{
"instrument": q.instrument_.instrument_id(),
"record_count": q.record_count_,
"latest_tstamp": q.latest_tstamp_.isoformat() if q.latest_tstamp_ else None,
"status": q.status_,
"reason": q.reason_,
}
)
return res
def pair_dicts(self) -> List[Dict[str, Any]]:
return [p.as_dict() for p in [pair.stats_history_[-1] for pair in self.pairs_]]
@@ -5,12 +5,12 @@ from typing import Any, Dict, List, Optional
import pandas as pd
# ---
from cvttpy_tools.base.base import NamedObject
from cvttpy_tools.base.app import App
from cvttpy_tools.base.config import Config
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.base.timeutils import NanosT, SecPerHour, current_nanoseconds, NanoPerSec, format_nanos_utc
from cvttpy_tools.base.logger import Log
from cvttpy_tools.timeutils import NanosT, SecPerHour, current_nanoseconds, NanoPerSec, format_nanos_utc
from cvttpy_tools.logger import Log
# ---
from cvttpy_trading.trading.instrument import ExchangeInstrument
@@ -8,7 +8,7 @@ from typing import Any, Dict, Optional, cast
import numpy as np
import pandas as pd
from cvttpy_tools.base.config import Config
from cvttpy_tools.config import Config
@dataclass
class DataWindowParams:
@@ -6,8 +6,8 @@ from typing import Any, Dict, List, Optional
import pandas as pd
# ---
from cvttpy_tools.base.base import NamedObject
from cvttpy_tools.base.config import Config
from cvttpy_tools.base import NamedObject
from cvttpy_tools.config import Config
from cvttpy_tools.settings.cvtt_types import JsonDictT
# ---
@@ -4,7 +4,7 @@ from abc import ABC, abstractmethod
from typing import Any, Dict, cast
# ---
from cvttpy_tools.base.config import Config
from cvttpy_tools.config import Config
# ---
from pairs_trading.lib.pt_strategy.prediction import Prediction
from pairs_trading.lib.pt_strategy.trading_pair import TradingPair
@@ -4,7 +4,7 @@ from typing import Any, Dict, List, Optional, Tuple
import pandas as pd
# ---
from cvttpy_tools.base.config import Config
from cvttpy_tools.config import Config
# ---
from cvttpy_trading.trading.instrument import ExchangeInstrument
# ---
@@ -5,7 +5,7 @@ from typing import Any, Dict, List, Optional, Tuple
import pandas as pd
# ---
from cvttpy_tools.base.config import Config
from cvttpy_tools.config import Config
# ---
from cvttpy_trading.trading.instrument import ExchangeInstrument
# ---
@@ -8,8 +8,8 @@ from typing import Any, Dict, List
import pandas as pd
# ---
from cvttpy_tools.base.base import NamedObject
from cvttpy_tools.base.config import Config
from cvttpy_tools.base import NamedObject
from cvttpy_tools.config import Config
# ---
from cvttpy_trading.trading.instrument import ExchangeInstrument
# ---
@@ -2,7 +2,7 @@ import hjson
from typing import Dict
from datetime import datetime
# ---
from cvttpy_tools.base.config import Config
from cvttpy_tools.config import Config
def load_config(config_path: str) -> Config:
@@ -2,7 +2,7 @@ import os
import glob
from typing import Dict, List, Tuple
# ---
from cvttpy_tools.base.config import Config
from cvttpy_tools.config import Config
# ---
from cvttpy_trading.trading.instrument import ExchangeInstrument
-1
View File
@@ -1 +0,0 @@
-354
View File
@@ -1,354 +0,0 @@
{
"cells": [
{
"cell_type": "markdown",
"id": "single-day-title",
"metadata": {},
"source": [
"# Single-Day Backtest Result Analysis\n",
"\n",
"This notebook analyzes the result of one single-day backtest stored in a SQLite database. Development is staged; Step 1 only selects the database file that later sections will read.\n",
"\n",
"Input assumptions for Step 1:\n",
"\n",
"- The default data directory is `data/` at the repository root.\n",
"- SQLite result files usually use `.db`, `.sqlite`, or `.sqlite3` extensions.\n",
"- The directory can be changed interactively if the result file lives elsewhere."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "imports-and-paths",
"metadata": {},
"outputs": [],
"source": [
"from pathlib import Path\n",
"import importlib\n",
"import sys\n",
"\n",
"from IPython.display import display\n",
"import ipywidgets as widgets\n",
"import pandas as pd\n",
"import panel as pn\n",
"\n",
"START_DIR = Path.cwd().resolve()\n",
"for candidate in (START_DIR, *START_DIR.parents):\n",
" if (candidate / \"scripts\" / \"spbt_day.py\").exists():\n",
" if str(candidate) not in sys.path:\n",
" sys.path.insert(0, str(candidate))\n",
" break\n",
"\n",
"import scripts.spbt_day as spbt_day\n",
"\n",
"spbt_day = importlib.reload(spbt_day)\n",
"pn.extension(\"tabulator\", \"plotly\")\n",
"\n",
"ANALYZE_BUTTON_COLUMN = spbt_day.ANALYZE_BUTTON_COLUMN\n",
"SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS = spbt_day.SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS\n",
"add_total_pnl = spbt_day.add_total_pnl\n",
"calculate_pair_theo_executions = spbt_day.calculate_pair_theo_executions\n",
"calculate_ranked_pairs_theo_ret = spbt_day.calculate_ranked_pairs_theo_ret\n",
"create_database_file_selector = spbt_day.create_database_file_selector\n",
"create_pair_theo_ret_analyze_grid = spbt_day.create_pair_theo_ret_analyze_grid\n",
"create_pair_trades_market_plot = spbt_day.create_pair_trades_market_plot\n",
"create_selected_pair_executions_grid = spbt_day.create_selected_pair_executions_grid\n",
"create_total_pnl_histogram = spbt_day.create_total_pnl_histogram\n",
"find_repo_root = spbt_day.find_repo_root\n",
"format_pair_name_for_display = spbt_day.format_pair_name_for_display\n",
"format_pair_names_for_display = spbt_day.format_pair_names_for_display\n",
"format_pair_theo_ret_for_analyze_grid = spbt_day.format_pair_theo_ret_for_analyze_grid\n",
"infer_trading_day_start_ns = spbt_day.infer_trading_day_start_ns\n",
"load_selector_pair_rankings = spbt_day.load_selector_pair_rankings\n",
"load_pair_market_data = spbt_day.load_pair_market_data\n",
"load_trading_instructions = spbt_day.load_trading_instructions\n",
"pair_name_from_analyze_event = spbt_day.pair_name_from_analyze_event\n",
"show_interactive_dataframe = spbt_day.show_interactive_dataframe\n",
"\n",
"REPO_ROOT = find_repo_root()\n",
"DEFAULT_DATA_DIR = REPO_ROOT / \"data\"\n",
"\n",
"REPO_ROOT, DEFAULT_DATA_DIR"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "database-file-selector",
"metadata": {},
"outputs": [],
"source": [
"db_selector = create_database_file_selector(\n",
" default_data_dir=DEFAULT_DATA_DIR,\n",
" repo_root=REPO_ROOT,\n",
")\n",
"\n",
"display(db_selector.widget)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "selected-database-helpers",
"metadata": {},
"outputs": [],
"source": [
"selected_database_path = db_selector.selected_database_path\n",
"connect_selected_database = db_selector.connect_selected_database\n",
"\n",
"# Later notebook sections can call selected_database_path() or connect_selected_database()."
]
},
{
"cell_type": "markdown",
"id": "selector-pair-rankings-context",
"metadata": {},
"source": [
"## Selector Pair Rankings\n",
"\n",
"Load `selector_pairs.pair_name` and `selector_pairs.mr_score` from the selected SQLite database. The JSON field `mr_score.final` is parsed as a numeric score and ranked descending with dense ranks, so tied scores share the same rank and the next distinct score gets the next rank.\n",
"\n",
"Rows with missing, malformed, non-numeric, or non-finite `mr_score.final` values are preserved, sorted after ranked rows, and marked in `mr_score_parse_status`."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "load-selector-pair-rankings",
"metadata": {},
"outputs": [],
"source": [
"conn = connect_selected_database()\n",
"try:\n",
" selector_pair_rankings = load_selector_pair_rankings(conn)\n",
"finally:\n",
" conn.close()\n",
"\n",
"selector_pair_rankings_display = format_pair_names_for_display(\n",
" selector_pair_rankings[[\"pair_rank\", \"pair_name\", \"mr_score_final\"]]\n",
")\n",
"with pd.option_context(\"display.max_rows\", None):\n",
" display(selector_pair_rankings_display)"
]
},
{
"cell_type": "markdown",
"id": "theoretical-return-context",
"metadata": {},
"source": [
"## Theoretical Return by Pair\n",
"\n",
"Load `trading_instructions` and calculate theoretical return for each ranked pair. Each pair starts from a fixed `$10,000` theoretical USD base. `TARGET` trades from the current theoretical position to the new target position, where target size is `10000 * strength / reference_price`; `CLOSE` liquidates the open position at the close row's `reference_price`; `HOLD` is ignored.\n",
"\n",
"`MIN_TARGET_STRENGTH_CHANGE_PCTG` can be raised above `0.0` to skip `TARGET` updates whose absolute percentage strength change is smaller than the threshold since the position was acquired. `num_trades` counts asset-level theoretical trades caused by effective `TARGET` and `CLOSE` rows. `realized_pnl` and `unrealized_pnl` are percentage returns relative to `$10,000`. The displayed dataframe is sorted by total return (`realized_pnl + unrealized_pnl`) ascending."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "target-change-threshold-input",
"metadata": {},
"outputs": [],
"source": [
"min_target_change_input = widgets.FloatText(\n",
" value=0.0,\n",
" description=\"Mininal TARGET change (%)\",\n",
" step=1.0,\n",
" layout=widgets.Layout(width=\"420px\"),\n",
" style={\"description_width\": \"190px\"},\n",
")\n",
"display(min_target_change_input)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "load-trading-instructions",
"metadata": {},
"outputs": [],
"source": [
"conn = connect_selected_database()\n",
"try:\n",
" trading_instructions = load_trading_instructions(conn)\n",
"finally:\n",
" conn.close()\n",
"\n",
"print(f\"Loaded {len(trading_instructions):,} trading instruction rows.\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "calculate-pair-theoretical-returns",
"metadata": {},
"outputs": [],
"source": [
"MIN_TARGET_STRENGTH_CHANGE_PCTG = float(min_target_change_input.value)\n",
"\n",
"pair_theo_ret = add_total_pnl(\n",
" calculate_ranked_pairs_theo_ret(\n",
" selector_pair_rankings,\n",
" trading_instructions,\n",
" min_pctg_change=MIN_TARGET_STRENGTH_CHANGE_PCTG,\n",
" )\n",
").sort_values(\n",
" [\"total_pnl\", \"pair_name\"],\n",
" ascending=[True, True],\n",
" kind=\"mergesort\",\n",
").drop(columns=\"total_pnl\").reset_index(drop=True)\n",
"\n",
"pair_theo_ret_display = format_pair_theo_ret_for_analyze_grid(pair_theo_ret)\n",
"pair_theo_ret_grid = create_pair_theo_ret_analyze_grid(\n",
" pair_theo_ret_display,\n",
" height=520,\n",
")\n",
"display(pair_theo_ret_grid)"
]
},
{
"cell_type": "markdown",
"id": "theoretical-return-histogram-context",
"metadata": {},
"source": [
"## Total Theoretical Return Distribution\n",
"\n",
"Plot the distribution of total theoretical return, calculated as `realized_pnl + unrealized_pnl`. Plotly chooses histogram bins automatically."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "plot-total-theoretical-return-histogram",
"metadata": {},
"outputs": [],
"source": [
"total_pnl_histogram = create_total_pnl_histogram(pair_theo_ret)\n",
"total_pnl_histogram_pane = pn.pane.Plotly(\n",
" total_pnl_histogram,\n",
" height=360,\n",
" sizing_mode=\"stretch_width\",\n",
")\n",
"\n",
"display(total_pnl_histogram_pane)"
]
},
{
"cell_type": "markdown",
"id": "individual-pair-analysis-context",
"metadata": {},
"source": [
"## Individual Pair Analysis\n",
"\n",
"Click the Analyze button in the Pair TheoRet grid to load detailed follow-up analysis for that row. The selected-pair execution table and market/trade chart are not calculated until an Analyze button is clicked."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "individual-pair-analysis",
"metadata": {},
"outputs": [],
"source": [
"selected_pair_name = None\n",
"selected_pair_theo_executions = pd.DataFrame()\n",
"selected_pair_theo_executions_display = pd.DataFrame(\n",
" columns=SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS\n",
")\n",
"selected_pair_market_data = pd.DataFrame()\n",
"selected_pair_market_trades_plot = None\n",
"\n",
"selected_pair_message = pn.pane.Markdown(\n",
" \"Click Analyze in the Pair TheoRet grid to load individual-pair details.\"\n",
")\n",
"selected_pair_theo_executions_grid = create_selected_pair_executions_grid(\n",
" selected_pair_theo_executions_display,\n",
" height=360,\n",
")\n",
"selected_pair_market_trades_plot_pane = pn.pane.Plotly(\n",
" None,\n",
" height=520,\n",
" sizing_mode=\"stretch_width\",\n",
")\n",
"\n",
"\n",
"def analyze_pair_click(event):\n",
" global selected_pair_name\n",
" global selected_pair_theo_executions\n",
" global selected_pair_theo_executions_display\n",
" global selected_pair_market_data\n",
" global selected_pair_market_trades_plot\n",
"\n",
" try:\n",
" selected_pair_name = pair_name_from_analyze_event(pair_theo_ret_grid, event)\n",
" selected_pair_message.object = (\n",
" f\"Selected pair: **{format_pair_name_for_display(selected_pair_name)}**\"\n",
" )\n",
"\n",
" selected_pair_theo_executions = calculate_pair_theo_executions(\n",
" selected_pair_name,\n",
" trading_instructions,\n",
" min_pctg_change=MIN_TARGET_STRENGTH_CHANGE_PCTG,\n",
" )\n",
" selected_pair_theo_executions_display = selected_pair_theo_executions.reindex(\n",
" columns=SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS\n",
" )\n",
" selected_pair_theo_executions_grid.value = selected_pair_theo_executions_display\n",
"\n",
" trading_day_start_ns = infer_trading_day_start_ns(trading_instructions)\n",
" conn = connect_selected_database()\n",
" try:\n",
" selected_pair_market_data = load_pair_market_data(\n",
" conn,\n",
" selected_pair_name,\n",
" trading_day_start_ns=trading_day_start_ns,\n",
" )\n",
" finally:\n",
" conn.close()\n",
"\n",
" selected_pair_market_trades_plot = create_pair_trades_market_plot(\n",
" selected_pair_name,\n",
" selected_pair_market_data,\n",
" selected_pair_theo_executions,\n",
" )\n",
" selected_pair_market_trades_plot_pane.object = selected_pair_market_trades_plot\n",
" except Exception as exc:\n",
" selected_pair_message.object = f\"**Error:** {exc}\"\n",
" selected_pair_market_trades_plot_pane.object = None\n",
"\n",
"\n",
"pair_theo_ret_grid.on_click(analyze_pair_click, column=ANALYZE_BUTTON_COLUMN)\n",
"\n",
"display(\n",
" pn.Column(\n",
" selected_pair_message,\n",
" \"### Theoretical Executions\",\n",
" selected_pair_theo_executions_grid,\n",
" \"### Trades on Market Data\",\n",
" selected_pair_market_trades_plot_pane,\n",
" )\n",
")"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "python3.12-venv (3.12.13.final.0)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.13"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
-295
View File
@@ -1,295 +0,0 @@
"""Panel application for single-day SPBT result analysis."""
from __future__ import annotations
from pathlib import Path
import sys
from typing import Any
import pandas as pd
import panel as pn
APP_DIR = Path(__file__).resolve().parent
REPO_ROOT = APP_DIR.parent
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
from scripts import spbt_day
pn.extension("tabulator", "plotly")
PAIR_THEO_RET_SORT_COLUMNS = ["total_pnl", "pair_name"]
PAIR_THEO_RET_DISPLAY_DROP_COLUMNS = ["total_pnl"]
APP_TITLE = "SPBT Day Analysis"
APP_ACCENT_COLOR = "#226c67"
APP_HEADER_COLOR = "#184c47"
class SpbtDayPanelApp:
"""Stateful Panel UI for single-day SPBT analysis."""
def __init__(self, repo_root: Path | None = None) -> None:
self.repo_root = (repo_root or spbt_day.find_repo_root(REPO_ROOT)).resolve()
self.selector_pair_rankings = pd.DataFrame()
self.trading_instructions = pd.DataFrame()
self.pair_theo_ret = pd.DataFrame()
self.selected_pair_theo_executions = pd.DataFrame()
self.selected_pair_name: str | None = None
self.min_pctg_change = 0.0
self.directory_input = pn.widgets.TextInput(
label="Directory",
value=str(self.repo_root / "data"),
)
self.show_all_files = pn.widgets.Checkbox(label="Show all files", value=False)
self.refresh_button = pn.widgets.Button(label="Refresh")
self.file_select = pn.widgets.Select(
label="SQLite result file",
options={},
width=360,
)
self.min_pctg_change_input = pn.widgets.FloatInput(
label="Mininal TARGET change (%)",
value=0.0,
step=1.0,
width=220,
)
self.calculate_button = pn.widgets.Button(
label="Calculate",
color="primary",
width=110,
)
self.status = pn.pane.Markdown("")
self.pair_theo_ret_table = spbt_day.create_pair_theo_ret_analyze_grid(
pd.DataFrame(),
height=420,
)
self.total_pnl_histogram = pn.pane.Plotly(
None,
height=360,
sizing_mode="stretch_width",
)
self.selected_pair_message = pn.pane.Markdown(
"Click Analyze in the Pair TheoRet grid to load individual-pair details."
)
self.selected_pair_executions_table = spbt_day.create_selected_pair_executions_grid(
height=320,
)
self.selected_pair_market_plot = pn.pane.Plotly(
None,
height=520,
sizing_mode="stretch_width",
)
self.refresh_button.on_click(self.refresh_files)
self.calculate_button.on_click(self.calculate)
self.directory_input.param.watch(self.refresh_files, "value")
self.show_all_files.param.watch(self.refresh_files, "value")
self.pair_theo_ret_table.on_click(
self.analyze_pair_click,
column=spbt_day.ANALYZE_BUTTON_COLUMN,
)
self.refresh_files()
def set_status(self, message: str, *, error: bool = False) -> None:
"""Update visible status text."""
prefix = "**Error:** " if error else ""
self.status.object = f"{prefix}{message}" if message else ""
def selected_database_path(self) -> Path:
"""Return the selected result database path."""
if not self.file_select.value:
raise ValueError("Select a SQLite result file before calculating.")
db_path = Path(str(self.file_select.value)).resolve()
if not db_path.exists():
raise FileNotFoundError(f"Selected database does not exist: {db_path}")
if not db_path.is_file():
raise ValueError(f"Selected database path is not a file: {db_path}")
return db_path
def refresh_files(self, *_events: Any) -> None:
"""Refresh selectable SQLite files from the configured directory."""
try:
directory = spbt_day.normalize_directory(
self.directory_input.value,
self.repo_root,
)
candidates = spbt_day.list_candidate_files(
directory,
show_all=self.show_all_files.value,
)
except Exception as exc:
self.file_select.options = {}
self.file_select.value = None
self.set_status(str(exc), error=True)
return
options = {path.name: str(path) for path in candidates}
previous_value = self.file_select.value
self.file_select.options = options
if previous_value in options.values():
self.file_select.value = previous_value
elif options:
self.file_select.value = next(iter(options.values()))
else:
self.file_select.value = None
if options:
self.set_status(f"Found {len(options):,} file(s) in {directory}.")
else:
self.set_status(f"No selectable files found in {directory}.")
def calculate(self, *_events: Any) -> None:
"""Load selected data and calculate all-pair TheoRet."""
self.calculate_button.loading = True
try:
db_path = self.selected_database_path()
self.min_pctg_change = float(self.min_pctg_change_input.value)
conn = spbt_day.connect_sqlite_read_only(db_path)
try:
self.selector_pair_rankings = spbt_day.load_selector_pair_rankings(conn)
self.trading_instructions = spbt_day.load_trading_instructions(conn)
finally:
conn.close()
self.pair_theo_ret = (
spbt_day.add_total_pnl(
spbt_day.calculate_ranked_pairs_theo_ret(
self.selector_pair_rankings,
self.trading_instructions,
min_pctg_change=self.min_pctg_change,
)
)
.sort_values(
PAIR_THEO_RET_SORT_COLUMNS,
ascending=[True, True],
kind="mergesort",
)
.drop(columns=PAIR_THEO_RET_DISPLAY_DROP_COLUMNS)
.reset_index(drop=True)
)
self.pair_theo_ret_table.value = spbt_day.format_pair_theo_ret_for_analyze_grid(
self.pair_theo_ret
)
self.total_pnl_histogram.object = spbt_day.create_total_pnl_histogram(
self.pair_theo_ret
)
self.clear_selected_pair_analysis()
self.set_status(
f"Calculated {len(self.pair_theo_ret):,} pair row(s) from {db_path.name}."
)
except Exception as exc:
self.set_status(str(exc), error=True)
finally:
self.calculate_button.loading = False
def clear_selected_pair_analysis(self) -> None:
"""Clear individual-pair outputs until a row Analyze button is clicked."""
self.selected_pair_name = None
self.selected_pair_theo_executions = pd.DataFrame()
self.selected_pair_message.object = (
"Click Analyze in the Pair TheoRet grid to load individual-pair details."
)
self.selected_pair_executions_table.value = pd.DataFrame(
columns=spbt_day.SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS
)
self.selected_pair_market_plot.object = None
def analyze_pair_click(self, event: Any) -> None:
"""Run selected-pair analysis from a Pair TheoRet Analyze button click."""
self.update_selected_pair(
spbt_day.pair_name_from_analyze_event(self.pair_theo_ret_table, event)
)
def analyze_pair_row(self, row: int) -> None:
"""Run selected-pair analysis for a Pair TheoRet table row."""
event = type("AnalyzeEvent", (), {"row": row})()
self.analyze_pair_click(event)
def update_selected_pair(self, pair_name: str) -> None:
"""Calculate selected-pair executions and market plot."""
if self.trading_instructions.empty:
self.clear_selected_pair_analysis()
return
self.selected_pair_name = pair_name
self.selected_pair_message.object = (
f"Selected pair: **{spbt_day.format_pair_name_for_display(pair_name)}**"
)
self.selected_pair_theo_executions = spbt_day.calculate_pair_theo_executions(
pair_name,
self.trading_instructions,
min_pctg_change=self.min_pctg_change,
)
self.selected_pair_executions_table.value = (
self.selected_pair_theo_executions.reindex(
columns=spbt_day.SELECTED_PAIR_EXECUTION_DISPLAY_COLUMNS
)
)
try:
trading_day_start_ns = spbt_day.infer_trading_day_start_ns(
self.trading_instructions
)
conn = spbt_day.connect_sqlite_read_only(self.selected_database_path())
try:
selected_pair_market_data = spbt_day.load_pair_market_data(
conn,
pair_name,
trading_day_start_ns=trading_day_start_ns,
)
finally:
conn.close()
self.selected_pair_market_plot.object = spbt_day.create_pair_trades_market_plot(
pair_name,
selected_pair_market_data,
self.selected_pair_theo_executions,
)
except Exception as exc:
self.selected_pair_market_plot.object = None
self.set_status(str(exc), error=True)
@property
def view(self) -> pn.template.FastListTemplate:
"""Return the app layout."""
controls = pn.Column(
"## Inputs",
pn.Row(self.directory_input, self.refresh_button),
self.show_all_files,
self.file_select,
self.min_pctg_change_input,
self.calculate_button,
self.status,
width=400,
)
main = pn.Column(
"## Pair TheoRet",
self.pair_theo_ret_table,
self.total_pnl_histogram,
"## Individual Pair",
self.selected_pair_message,
"### Theoretical Executions",
self.selected_pair_executions_table,
"### Trades on Market Data",
self.selected_pair_market_plot,
)
return pn.template.FastListTemplate(
title=APP_TITLE,
sidebar=[controls],
main=[main],
sidebar_width=430,
accent_base_color=APP_ACCENT_COLOR,
header_background=APP_HEADER_COLOR,
main_layout=None,
)
app_controller = SpbtDayPanelApp()
app = app_controller.view
app.servable(title=APP_TITLE)
+201 -13
View File
@@ -1,13 +1,201 @@
# Interactive analysis
ipykernel>=6.29,<7
ipywidgets>=8.1,<9
itables>=2.2,<3
jupyter>=1.1,<2
nbformat>=5.10,<6
pandas>=2.2,<3
panel>=1.5,<2
plotly>=5.24,<7
# Verification
nbmake>=1.5,<2
pytest>=8,<9
aiohttp>=3.8.4
aiosignal>=1.3.1
async-timeout>=4.0.2
attrs>=21.2.0
beautifulsoup4>=4.10.0
black>=23.3.0
flake8>=6.0.0
certifi>=2020.6.20
chardet>=4.0.0
charset-normalizer>=3.1.0
click>=8.0.3
colorama>=0.4.4
configobj>=5.0.6
cryptography>=3.4.8
distro>=1.7.0
docker>=5.0.3
dockerpty>=0.4.1
docopt>=0.6.2
eyeD3>=0.8.10
filelock>=3.6.0
frozenlist>=1.3.3
grpcio>=1.30.2
hjson>=3.0.2
html5lib>=1.1
httplib2>=0.20.2
idna>=3.3
ipython>=8.18.1
ipywidgets>=8.1.1
ifaddr>=0.1.7
IMDbPY>=2021.4.18
ipykernel>=6.29.5
jeepney>=0.7.1
jsonschema>=3.2.0
jupyter>=1.0.0
keyring>=23.5.0
launchpadlib>=1.10.16
lazr.restfulclient>=0.14.4
lazr.uri>=1.0.6
lxml>=4.8.0
Mako>=1.1.3
Markdown>=3.3.6
MarkupSafe>=2.0.1
matplotlib>=3.10.3
more-itertools>=8.10.0
multidict>=6.0.4
mypy>=0.942
mypy-extensions>=0.4.3
nbformat>=5.10.2
netaddr>=0.8.0
######### netifaces>=0.11.0
numpy>=1.26.4,<2.3.0
oauthlib>=3.2.0
packaging>=23.1
pandas>=2.2.3
pathspec>=0.11.1
pexpect>=4.8.0
Pillow>=9.0.1
platformdirs>=3.2.0
plotly>=5.19.0
protobuf>=3.12.4
psutil>=5.9.0
ptyprocess>=0.7.0
pycurl>=7.44.1
pyelftools>=0.27
Pygments>=2.11.2
pyparsing>=2.4.7
pyrsistent>=0.18.1
python-debian>=0.1.43 #+ubuntu1.1
python-dotenv>=0.19.2
python-magic>=0.4.24
python-xlib>=0.29
pyxdg>=0.27
PyYAML>=6.0
reportlab>=3.6.8
requests>=2.25.1
requests-file>=1.5.1
scipy<1.13.0
seaborn>=0.13.2
SecretStorage>=3.3.1
setproctitle>=1.2.2
simpleeval>=1.0.3
six>=1.16.0
soupsieve>=2.3.1
ssh-import-id>=5.11
statsmodels>=0.14.4
texttable>=1.6.4
tldextract>=3.1.2
tomli>=1.2.2
######## typed-ast>=1.4.3
types-aiofiles>=0.1
types-annoy>=1.17
types-appdirs>=1.4
types-atomicwrites>=1.4
types-aws-xray-sdk>=2.8
types-babel>=2.9
types-backports-abc>=0.5
types-backports.ssl-match-hostname>=3.7
types-beautifulsoup4>=4.10
types-bleach>=4.1
types-boto>=2.49
types-braintree>=4.11
types-cachetools>=4.2
types-caldav>=0.8
types-certifi>=2020.4
types-characteristic>=14.3
types-chardet>=4.0
types-click>=7.1
types-click-spinner>=0.1
types-colorama>=0.4
types-commonmark>=0.9
types-contextvars>=0.1
types-croniter>=1.0
types-cryptography>=3.3
types-dataclasses>=0.1
types-dateparser>=1.0
types-DateTimeRange>=0.1
types-decorator>=0.1
types-Deprecated>=1.2
types-docopt>=0.6
types-docutils>=0.17
types-editdistance>=0.5
types-emoji>=1.2
types-entrypoints>=0.3
types-enum34>=1.1
types-filelock>=3.2
types-first>=2.0
types-Flask>=1.1
types-freezegun>=1.1
types-frozendict>=0.1
types-futures>=3.3
types-html5lib>=1.1
types-httplib2>=0.19
types-humanfriendly>=9.2
types-ipaddress>=1.0
types-itsdangerous>=1.1
types-JACK-Client>=0.1
types-Jinja2>=2.11
types-jmespath>=0.10
types-jsonschema>=3.2
types-Markdown>=3.3
types-MarkupSafe>=1.1
types-mock>=4.0
types-mypy-extensions>=0.4
types-mysqlclient>=2.0
types-oauthlib>=3.1
types-orjson>=3.6
types-paramiko>=2.7
types-Pillow>=8.3
types-polib>=1.1
types-prettytable>=2.1
types-protobuf>=3.17
types-psutil>=5.8
types-psycopg2>=2.9
types-pyaudio>=0.2
types-pycurl>=0.1
types-pyfarmhash>=0.2
types-Pygments>=2.9
types-PyMySQL>=1.0
types-pyOpenSSL>=20.0
types-pyRFC3339>=0.1
types-pysftp>=0.2
types-pytest-lazy-fixture>=0.6
types-python-dateutil>=2.8
types-python-gflags>=3.1
types-python-nmap>=0.6
types-python-slugify>=5.0
types-pytz>=2021.1
types-pyvmomi>=7.0
types-PyYAML>=5.4
types-redis>=3.5
types-requests>=2.25
types-retry>=0.9
types-seaborn>0.13.2
types-selenium>=3.141
types-Send2Trash>=1.8
types-setuptools>=57.4
types-simplejson>=3.17
types-singledispatch>=3.7
types-six>=1.16
types-slumber>=0.7
types-stripe>=2.59
types-tabulate>=0.8
types-termcolor>=1.1
types-toml>=0.10
types-toposort>=1.6
types-ttkthemes>=3.2
types-typed-ast>=1.4
types-tzlocal>=0.1
types-ujson>=0.1
types-vobject>=0.9
types-waitress>=0.1
types-Werkzeug>=1.0
types-xxhash>=2.0
typing-extensions>=3.10.0.2
Unidecode>=1.3.3
urllib3>=1.26.5
wadllib>=1.3.6
webencodings>=0.5.1
websocket-client>=1.2.3
yarl>=1.9.1
zipp>=1.0.0
@@ -4,9 +4,9 @@ import os
from typing import Any, Dict, List, Tuple
# ---
from cvttpy_tools.base.app import App
from cvttpy_tools.base.base import NamedObject
from cvttpy_tools.base.config import CvttAppConfig
from cvttpy_tools.app import App
from cvttpy_tools.base import NamedObject
from cvttpy_tools.config import CvttAppConfig
# ---
from cvttpy_trading.trading.instrument import ExchangeInstrument
@@ -182,7 +182,7 @@
"import os\n",
"import importlib\n",
"\n",
"from cvttpy_tools.base.config import Config\n",
"from cvttpy_tools.config import Config\n",
"\n",
"def load_config_from_file() -> Optional[Dict]:\n",
" global DB_TABLE_NAME\n",
-1
View File
@@ -1 +0,0 @@
-1
View File
@@ -1 +0,0 @@
-7
View File
@@ -1,7 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$repo_root"
panel serve panel/spbt_day_panel.py --show "$@"
-1262
View File
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
-218
View File
@@ -1,218 +0,0 @@
import importlib.util
import sqlite3
from pathlib import Path
import pandas as pd
def load_panel_app_module():
module_path = Path("panel/spbt_day_panel.py").resolve()
spec = importlib.util.spec_from_file_location("spbt_day_panel_app", module_path)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
return module
def create_panel_fixture_db(db_path: Path) -> None:
trading_day_start_ns = pd.Timestamp("2026-06-17T00:00:00Z").value
conn = sqlite3.connect(db_path)
try:
conn.execute(
"""
CREATE TABLE selector_pairs (
time_ns INTEGER,
tstamp TEXT,
pair_name TEXT,
instrument_a TEXT,
instrument_b TEXT,
mr_score TEXT
)
"""
)
conn.execute(
"""
CREATE TABLE trading_instructions (
time_ns INTEGER,
tstamp TEXT,
book_id TEXT,
strategy_id TEXT,
type TEXT,
data TEXT
)
"""
)
conn.execute(
"""
CREATE TABLE ohlcv_1min (
tstamp TEXT,
tstamp_ns INTEGER,
exch_acct TEXT,
exchange_id TEXT,
instrument_id TEXT,
interval_sec INTEGER,
open REAL,
high REAL,
low REAL,
close REAL,
volume REAL,
vwap REAL,
num_trades INTEGER
)
"""
)
conn.execute(
"INSERT INTO selector_pairs VALUES (?, ?, ?, ?, ?, ?)",
(
10,
"2026-06-17T00:00:00Z",
"AAA:USD-BBB:USD",
"EXCH:PAIR-AAA-USD",
"EXCH:PAIR-BBB-USD",
'{"final":"0.5"}',
),
)
conn.executemany(
"INSERT INTO trading_instructions VALUES (?, ?, ?, ?, ?, ?)",
[
(
trading_day_start_ns,
"2026-06-17T00:00:00Z",
"book",
"strategy-AAA:USD-BBB:USD",
"TARGET_POSITION",
(
'{"action":"TARGET","quote_asset":"USD","assets":'
'{"AAA":{"reference_price":"100","strength":"0.5"},'
'"BBB":{"reference_price":"50","strength":"-0.5"}}}'
),
),
(
trading_day_start_ns + 60_000_000_000,
"2026-06-17T00:01:00Z",
"book",
"strategy-AAA:USD-BBB:USD",
"CLOSE_POSITION",
(
'{"action":"CLOSE","quote_asset":"USD","assets":'
'{"AAA":{"reference_price":"110"},'
'"BBB":{"reference_price":"45"}}}'
),
),
],
)
conn.executemany(
"INSERT INTO ohlcv_1min VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
[
(
"2026-06-17T00:00:00Z",
trading_day_start_ns,
"EXCH",
"EXCH",
"PAIR-AAA-USD",
60,
100.0,
100.0,
100.0,
100.0,
1.0,
100.0,
1,
),
(
"2026-06-17T00:00:00Z",
trading_day_start_ns,
"EXCH",
"EXCH",
"PAIR-BBB-USD",
60,
50.0,
50.0,
50.0,
50.0,
1.0,
50.0,
1,
),
],
)
conn.commit()
finally:
conn.close()
def test_pair_analyze_grid_keeps_clean_labels_and_full_pair_values():
module = load_panel_app_module()
pair_theo_ret = pd.DataFrame(
{
"pair_name": ["BTC:USD-ETH:USD", "ADA:USD-BTC:USD"],
"mr_ranking": [2, 1],
"realized_pnl": [0.0, 0.0],
"unrealized_pnl": [0.0, 0.0],
}
)
formatted = module.spbt_day.format_pair_theo_ret_for_analyze_grid(pair_theo_ret)
assert formatted["pair_name"].tolist() == ["BTC-ETH", "ADA-BTC"]
assert formatted[module.spbt_day.PAIR_NAME_VALUE_COLUMN].tolist() == [
"BTC:USD-ETH:USD",
"ADA:USD-BTC:USD",
]
def test_panel_app_uses_fast_list_template(tmp_path):
module = load_panel_app_module()
app = module.SpbtDayPanelApp(repo_root=tmp_path)
view = app.view
assert isinstance(view, module.pn.template.FastListTemplate)
assert view.title == module.APP_TITLE
assert view.sidebar_width == 430
assert view.accent_base_color == module.APP_ACCENT_COLOR
assert view.header_background == module.APP_HEADER_COLOR
assert len(view.sidebar) == 1
assert len(view.main) == 1
def test_panel_app_calculates_pairs_and_selected_pair_outputs(tmp_path):
module = load_panel_app_module()
data_dir = tmp_path / "data"
data_dir.mkdir()
db_path = data_dir / "20260617.spbt_results.db"
create_panel_fixture_db(db_path)
app = module.SpbtDayPanelApp(repo_root=tmp_path)
app.directory_input.value = str(data_dir)
app.refresh_files()
app.min_pctg_change_input.value = 0.0
app.calculate()
assert app.file_select.value == str(db_path)
assert app.file_select.width == 360
assert app.min_pctg_change_input.width == 220
assert app.calculate_button.width == 110
assert app.total_pnl_histogram.sizing_mode == "stretch_width"
assert app.selected_pair_market_plot.sizing_mode == "stretch_width"
assert app.pair_theo_ret_table.pagination is None
assert app.pair_theo_ret_table.layout == "fit_data_table"
assert app.pair_theo_ret_table.value["pair_name"].tolist() == ["AAA-BBB"]
assert (
app.pair_theo_ret_table.value[module.spbt_day.PAIR_NAME_VALUE_COLUMN].tolist()
== ["AAA:USD-BBB:USD"]
)
assert app.selected_pair_name is None
assert app.selected_pair_executions_table.value.empty
assert app.selected_pair_market_plot.object is None
app.analyze_pair_row(0)
assert app.selected_pair_name == "AAA:USD-BBB:USD"
assert app.selected_pair_executions_table.value["action"].tolist() == [
"TARGET",
"TARGET",
"CLOSE",
"CLOSE",
]
assert app.selected_pair_market_plot.object is not None