gitadmin/fxhnt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(bybit): paper-book backfill — persist the live book daily over full history (multi-year record like binance)

Browse Source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-06-25 21:38:26 +02:00
parent c48c0dba95
commit 8eb23f3878
4 changed files with 599 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
src/fxhnt/application/bybit_paper_backfill.py Normal file
View File

@@ -0,0 +1,131 @@
"""Bybit LIVE paper-book BACKFILL — walk the full Bybit history and persist the live paper book daily, so
`/paper?venue=bybit` shows a multi-year record exactly like the (already-backfilled) Binance book.
The Binance equivalent is `paper_backfill.backfill_paper_book` (driven by the `paper-backfill` CLI). This is
the Bybit analogue: instead of recomputing per-date sleeve weights and running the risk overlay, it reuses
the SAME seams the LIVE Bybit derive (`bybit_paper_book.derive_and_persist_bybit_paper_book`) uses, but for
EVERY historical day rather than only the latest:
* `bybit_paper_weights.sleeve_weights_and_contributions` — per-sleeve, per-day, per-symbol `(weight,
return)` pairs, reconciled to each edge's GROSS return engine (the HARD reconciliation invariant). The
runner emits ALL days in one call, so the backfill extracts the whole history per sleeve ONCE (no
per-date engine re-run).
* naive EQUAL-WEIGHT 4-edge combination (`combined_weight = (1/n)·sleeve_weight`) — the DEPLOY book, the
same combination the live derive uses (the A/B proved naive eq-wt is best OOS for the Bybit book; the
live overlay is intentionally NOT applied here, matching the live book + forward track's design).
* `paper_book.persist_paper_book` — sizes target positions (`weight·equity/price` at THAT day's close),
diffs vs the PRIOR day's bybit book → trades, charges the 5.5bp per-turnover cost (the SAME haircut the
live book + forward track charge, so the backfill + forward record form ONE continuous book), and
compounds the book MTM into nav. Per-`run_date` idempotent replace → re-running is safe.
Dates are processed ASCENDING so the compounding nav equity (`nav_equity_before` + prior-book MTM) builds
date-by-date. Everything is written under venue="bybit" (`PaperRepo(venue="bybit")`), so the Binance book is
untouched. READ-ONLY against the warehouse store; only the paper store is written. Memory-bounded: the weight
extraction reuses the `_UniverseRestrictedStore` seam, so only the liquid `universe` symbols ever load.
CONTINUITY: the LAST backfilled day == what the single-day live derive
(`derive_and_persist_bybit_paper_book`) produces — both take the latest day's `combined_symbol_weights` and
push it through `persist_paper_book`. The backfill simply fills every PRIOR day with the same machinery.
"""
from __future__ import annotations
import datetime as dt
from typing import Any
from fxhnt.adapters.persistence.paper_repo import PaperRepo
from fxhnt.application.bybit_book_eval import _DEFAULT_BYBIT_SLEEVES, _bounded_store
from fxhnt.application.bybit_paper_book import _BYBIT_COST_RATE, _epoch_day_to_iso
from fxhnt.application.bybit_paper_weights import sleeve_weights_and_contributions
from fxhnt.application.paper_book import persist_paper_book
def combined_symbol_weights_by_day(
store: Any, *, universe: set[str] | None = None,
unlock_events: list[dict[str, Any]] | None = None,
sleeves: tuple[str, ...] = _DEFAULT_BYBIT_SLEEVES,
) -> dict[int, dict[str, dict[str, float]]]:
"""`{epoch_day: {sleeve: {symbol: combined_weight}}}` over the FULL history of every sleeve, combined
NAIVE EQUAL-WEIGHT (`combined_weight = (1/n)·sleeve_weight`).
This is the historical counterpart of `bybit_paper_book.combined_symbol_weights` (which returns only the
latest day): each sleeve's per-day per-symbol `(weight, return)` pairs are pulled ONCE via
`sleeve_weights_and_contributions` (the reconciled extraction — Σ w·r == the edge's GROSS return), then
every day on which a sleeve has weights contributes that sleeve's `(1/n)`-scaled weights to that day.
A sleeve with no weights (no data / uncomputable, e.g. unlock with no calendar) is simply absent on every
day. A day on which NO sleeve has weights is absent from the result. Zero-weight symbols are dropped (a
zero target adds nothing to the book). Pure given the store; READ-ONLY; memory-bounded (the runners read
only the `universe` subset)."""
n = len(sleeves)
if n == 0:
return {}
by_day: dict[int, dict[str, dict[str, float]]] = {}
for sleeve in sleeves:
per_day = sleeve_weights_and_contributions(store, sleeve, universe=universe,
unlock_events=unlock_events)
for day, row in per_day.items():
scaled = {sym: (1.0 / n) * w for sym, (w, _r) in row.items() if w != 0.0}
if scaled:
by_day.setdefault(day, {})[sleeve] = scaled
return by_day
def closes_by_day(store: Any, *, universe: set[str] | None = None) -> dict[int, dict[str, float]]:
"""`{epoch_day: {symbol: close}}` from the Bybit `close` panel — that day's mark for every symbol. Reuses
the universe-restricted read so only the liquid symbols load. READ-ONLY. (The live derive marks only the
latest close via `latest_prices`; the backfill needs each day's close to size + MTM that day.)"""
panel = _bounded_store(store, universe).crypto_close_panel()
out: dict[int, dict[str, float]] = {}
for sym, series in panel.items():
for day, close in series.items():
out.setdefault(int(day), {})[sym] = float(close)
return out
def backfill_bybit_paper_book(
repo: PaperRepo, store: Any, *, capital: float, at: dt.datetime,
universe: set[str] | None = None,
unlock_events: list[dict[str, Any]] | None = None,
cost_rate: float = _BYBIT_COST_RATE,
sleeves: tuple[str, ...] = _DEFAULT_BYBIT_SLEEVES,
since_day: int | None = None,
) -> int:
"""Walk every historical Bybit day on which the 4-edge weights exist and persist the live eq-wt book for
each day under venue="bybit". Returns the number of dates with a persisted nav row.
`repo` MUST be `PaperRepo(venue="bybit")`. For each day ascending:
1. the naive eq-wt 4-edge per-symbol target weights for THAT day (from
`combined_symbol_weights_by_day`);
2. that day's Bybit close per symbol = the mark;
3. `persist_paper_book` sizes target positions (`weight·equity/price`), diffs vs the PRIOR bybit book →
trades, charges the 5.5bp per-turnover cost, and compounds the prior book MTM into equity — written
under venue="bybit" (idempotent per run_date).
Ascending order is REQUIRED so the compounding nav (`nav_equity_before` + prior-book MTM) and the
prior-day diff are causal. Per-`run_date` idempotent replace makes re-running safe (each date's
positions/trades/nav are rewritten, never duplicated). `since_day` (epoch day) restricts to days STRICTLY
after it — the resumable / incremental fill (the days before are left as already-persisted history).
READ-ONLY against the warehouse `store`; only the paper store is written. The shared `_BYBIT_COST_RATE`
keeps the backfill and the live forward track on the SAME cost basis → one continuous book."""
weights_by_day = combined_symbol_weights_by_day(store, universe=universe,
unlock_events=unlock_events, sleeves=sleeves)
if not weights_by_day:
return 0
closes = closes_by_day(store, universe=universe)
days = sorted(d for d in weights_by_day if since_day is None or d > since_day)
persisted = 0
for day in days:
combined = weights_by_day[day]
prices = closes.get(day, {})
run_date = _epoch_day_to_iso(day)
# The per-symbol weights already carry the 1/n share, so each sleeve's sleeve_weight is 1.0 and the
# cost rate is the flat Bybit 5.5bp — IDENTICAL to the live derive's call into persist_paper_book.
sleeve_weights = {sleeve: 1.0 for sleeve in combined}
cost_rates = {sleeve: cost_rate for sleeve in combined}
persist_paper_book(
repo, capital=capital, run_date=run_date, sleeve_weights=sleeve_weights,
symbol_weights_by_sleeve=combined, prices=prices, at=at, enabled=True,
leverage=1.0, cost_rates=cost_rates)
persisted += 1
return persisted

73
src/fxhnt/cli.py
View File

@@ -2913,6 +2913,79 @@ def bybit_paper_book_cmd(
f"View at /paper?venue=bybit.")
@app.command("bybit-paper-backfill")
def bybit_paper_backfill_cmd(
min_dollar_vol: float = typer.Option(
10_000_000.0, "--min-dollar-vol",
help="Liquid-universe floor (median daily turnover, USD): only this liquid subset of bybit_features "
"is READ, which keeps the per-symbol weight extraction memory-bounded (the measured-optimal "
"default)."),
capital: float | None = typer.Option(
None, "--capital",
help="Starting capital for the backfilled book (default: settings.paper_capital)."),
rebuild: bool = typer.Option(
False, "--rebuild",
help="full rebuild of the whole Bybit history (the rare path). Default is INCREMENTAL: fill only "
"days STRICTLY AFTER the last persisted venue=bybit paper_nav date (resumable — the backfill "
"is idempotent per run_date, so resuming from the last date is exact)."),
) -> None:
"""BACKFILL the Bybit LIVE paper book over its FULL history (own-memory; the heavy ~1600-day counterpart
of the single-day `bybit-paper-book` asset): for EVERY historical day on which the 4-edge weights exist,
derive the naive-eq-wt 4-edge per-symbol book on THAT day's Bybit close, diff vs the prior bybit day →
trades, charge the 5.5bp turnover (the SAME haircut as the live book + forward track, so backfill +
forward form ONE continuous book), and compound the book MTM into nav — all written under venue="bybit".
The cockpit then shows a multi-year record at `/paper?venue=bybit`, just like the Binance book.
READ-ONLY against the warehouse; only venue="bybit" paper rows are written (the Binance book is
untouched). IDEMPOTENT per run_date (re-runnable). Default INCREMENTAL: resumes after the last persisted
bybit nav date; `--rebuild` recomputes the whole history."""
import datetime as dt
import logging
from fxhnt.adapters.orchestration.assets import _data_dir
from fxhnt.adapters.persistence.paper_repo import PaperRepo
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
from fxhnt.application.bybit_liquidity import liquid_universe
from fxhnt.application.bybit_paper_backfill import backfill_bybit_paper_book
from fxhnt.application.bybit_paper_book import BYBIT_VENUE
from fxhnt.application.unlock_calendar_loader import load_unlock_events, operational_unlock_repo
log = logging.getLogger(__name__)
settings = get_settings()
cap = settings.paper_capital if capital is None else capital
store = TimescaleFeatureStore(settings.operational_dsn, table="bybit_features")
try:
universe = liquid_universe(store, min_dollar_vol=min_dollar_vol) or None
try:
unlock_events = load_unlock_events(operational_unlock_repo(settings),
f"{_data_dir()}/unlock_calendar.json")
except Exception as exc: # noqa: BLE001 — absent calendar → unlock sleeve omitted, not a crash
log.warning("bybit-paper-backfill: could not load unlock calendar: %s", exc)
unlock_events = []
repo = PaperRepo(settings.operational_dsn, venue=BYBIT_VENUE)
repo.migrate()
# INCREMENTAL (default): resume STRICTLY AFTER the last persisted bybit nav date. The per-day
# computation depends only on strictly-prior state (positions/nav are strictly-before reads), so
# resuming reproduces exactly what a full rebuild would for the new days. --rebuild forces the whole
# history (first run / rare recompute).
since_day: int | None = None
if not rebuild:
last_done = repo.latest_nav_run_date()
if last_done is not None:
since_day = (dt.date.fromisoformat(last_done) - dt.date(1970, 1, 1)).days
n = backfill_bybit_paper_book(
repo, store, capital=cap, at=dt.datetime.now(dt.UTC),
universe=universe, unlock_events=unlock_events, since_day=since_day)
finally:
store.close()
scope = f"{len(universe)} liquid symbols" if universe else "full universe (no liquid subset)"
mode = "rebuild" if rebuild else "incremental"
typer.echo(f"bybit-paper-backfill: persisted {n} day(s) ({mode}) from {scope} "
f"(min_dollar_vol=${min_dollar_vol:,.0f}, capital=${cap:,.0f}). "
f"View the multi-year record at /paper?venue=bybit.")
@app.command("bybit-overlay-ab")
def bybit_overlay_ab(
liquid: bool = typer.Option(

303
tests/integration/test_bybit_paper_backfill.py Normal file
View File

@@ -0,0 +1,303 @@
"""Bybit LIVE paper-book BACKFILL: walk the full Bybit history and persist the live eq-wt 4-edge book daily
under venue="bybit", so `/paper?venue=bybit` shows a multi-year record like the Binance book. NO network
(in-memory TimescaleFeatureStore + sqlite PaperRepo).
Proves:
* the backfill walks ALL weight days and persists a nav point per day (N days → N nav rows, ascending);
* CONTINUITY/RECONCILIATION — each backfilled day's book is the eq-wt 4-edge book (reuses the reconciled
weight seam), and the LAST backfilled day == what the single-day live derive produces;
* IDEMPOTENT — running twice yields the same rows (per-date replace), and does NOT touch venue="binance";
* READ-ONLY against the warehouse (WriteTripwire) + memory-bounded (universe-restricted reads).
"""
from __future__ import annotations
import datetime as dt
import math
from fxhnt.adapters.persistence.paper_repo import PaperRepo
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
from fxhnt.application.bybit_paper_backfill import (
backfill_bybit_paper_book,
closes_by_day,
combined_symbol_weights_by_day,
)
from fxhnt.application.bybit_paper_book import (
BYBIT_VENUE,
combined_symbol_weights,
derive_and_persist_bybit_paper_book,
)
_DAY = 86_400
_AT = dt.datetime(2026, 6, 24, 23, 30, tzinfo=dt.UTC)
_CAPITAL = 100_000.0
def _seed(store: TimescaleFeatureStore, *, days: int = 200) -> list[dict]:
"""Seed BTC + 3 alts with close/funding/spot/long_ratio over `days` — enough history for tstrend, funding/
spot for xsfunding, oscillating long_ratio for positioning, and one unlock cliff. Mirrors the live-book +
reconciliation seeds so the backfill reconciles to the same engines."""
import math as _m
syms = ["BTCUSDT", "AAAUSDT", "BBBUSDT", "CCCUSDT"]
px = {s: 100.0 for s in syms}
px["BTCUSDT"] = 50_000.0
for d in range(days):
crowd = (d % 3 == 0)
for s in syms:
drift = {"BTCUSDT": 0.001, "AAAUSDT": 0.002, "BBBUSDT": -0.001, "CCCUSDT": 0.0015}[s]
wobble = 0.003 * _m.sin(d / 5.0 + hash(s) % 7)
px[s] *= (1.0 + drift + wobble)
funding = 0.0005 if s != "BTCUSDT" else 0.0001
spot = px[s] * (1.0 + 0.0002)
ratio = (0.70 if crowd else 0.30) if s == "AAAUSDT" else \
(0.30 if crowd else 0.70) if s == "BBBUSDT" else 0.50
store.write_features(s, [(d * _DAY, {
"close": px[s], "funding": funding, "spot_close": spot, "long_ratio": ratio})])
return [{"sym": "AAA", "cliff_day": days - 5, "n": 5_000_000.0, "cat": "insiders"}]
class _WriteTripwireStore:
"""Read-only proxy: any warehouse write attempt is a hard failure. The backfill must only READ the
warehouse (it writes ONLY the paper store, which is a separate object)."""
_FORBIDDEN = frozenset({
"write_features", "write_features_bulk", "upsert_feature_rows", "upsert_membership",
"_create_schema", "_bulk_upsert",
})
def __init__(self, inner: TimescaleFeatureStore) -> None:
object.__setattr__(self, "_inner", inner)
def __getattr__(self, name: str):
if name in _WriteTripwireStore._FORBIDDEN:
raise AssertionError(f"READ-ONLY violation: backfill called warehouse write method {name!r}")
return getattr(self._inner, name)
def test_backfill_persists_a_nav_point_per_weight_day_ascending() -> None:
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
events = _seed(store)
repo = PaperRepo("sqlite://", venue=BYBIT_VENUE)
repo.migrate()
n = backfill_bybit_paper_book(repo, store, capital=_CAPITAL, at=_AT, unlock_events=events)
store.close()
# Every day the eq-wt weights exist gets a nav row.
hist = repo.nav_history()
assert n == len(hist) > 1, "backfill should persist a multi-day nav series"
# Ascending, one point per day, no dupes.
dates = [p.run_date for p in hist]
assert dates == sorted(dates)
assert len(dates) == len(set(dates))
def test_backfill_day_count_matches_the_combined_weight_days() -> None:
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
events = _seed(store)
weights_by_day = combined_symbol_weights_by_day(store, unlock_events=events)
repo = PaperRepo("sqlite://", venue=BYBIT_VENUE)
repo.migrate()
n = backfill_bybit_paper_book(repo, store, capital=_CAPITAL, at=_AT, unlock_events=events)
store.close()
assert n == len(weights_by_day)
assert {p.run_date for p in repo.nav_history()} == {
(dt.date(1970, 1, 1) + dt.timedelta(days=d)).isoformat() for d in weights_by_day}
def test_last_backfilled_day_matches_the_single_day_live_derive_weights() -> None:
"""CONTINUITY: the backfill and the single-day live derive share the SAME weight seam, so the LAST
backfilled day reproduces the live derive's per-symbol WEIGHTS for the sleeves that fall on the global
latest day. (Both call `combined_symbol_weights*` → `persist_paper_book`. The qty differs only by the
equity base — the backfill has compounded the full history while the live derive seeds at `capital` — so
the load-bearing continuity invariant is the per-symbol WEIGHT, which is what makes backfill + forward
ONE book. NOTE: the live snapshot BUNDLES each sleeve's OWN latest day onto the max run_date, while the
backfill places each sleeve on its TRUE day; so the comparison is over the sleeves whose own latest day
IS the global latest day — for those the two agree exactly.)"""
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
events = _seed(store)
# The live derive's combined weights for the latest (bundled) run_date.
live_combined, latest_day = combined_symbol_weights(store, unlock_events=events)
assert latest_day is not None
latest_iso = (dt.date(1970, 1, 1) + dt.timedelta(days=latest_day)).isoformat()
# The backfill's combined weights for that SAME day (true per-day, no bundling).
bf_by_day = combined_symbol_weights_by_day(store, unlock_events=events)
bf_combined = bf_by_day[latest_day]
# The sleeves that genuinely fall on the global latest day appear in BOTH with identical per-symbol
# weights (the shared reconciled seam). Bundled-stale sleeves (own latest day < global) are in the live
# snapshot only — by design.
common = set(live_combined) & set(bf_combined)
assert common, "no sleeve on the global latest day (seed too short?)"
for sleeve in common:
assert bf_combined[sleeve].keys() == live_combined[sleeve].keys()
for sym, w in live_combined[sleeve].items():
assert math.isclose(bf_combined[sleeve][sym], w, rel_tol=1e-12, abs_tol=1e-15)
# And the backfill actually persists a book on that latest day (the forward track continues from here).
live_repo = PaperRepo("sqlite://", venue=BYBIT_VENUE)
live_repo.migrate()
derive_and_persist_bybit_paper_book(live_repo, store, capital=_CAPITAL, at=_AT, unlock_events=events)
bf_repo = PaperRepo("sqlite://", venue=BYBIT_VENUE)
bf_repo.migrate()
backfill_bybit_paper_book(bf_repo, store, capital=_CAPITAL, at=_AT, unlock_events=events)
store.close()
assert bf_repo.positions_at(latest_iso), "backfill must persist a book on the latest day"
# The backfill's latest persisted nav date IS the live derive's run_date — the forward track resumes here.
assert bf_repo.latest_nav_run_date() == live_repo.latest_nav_run_date() == latest_iso
def test_backfill_per_day_book_reconciles_to_eqwt_edge_return() -> None:
"""RECONCILIATION: the backfilled book's per-symbol weights on each day ARE the reconciled eq-wt edge
weights (Σ_symbol weight·return == the GROSS edge return / n, summed over sleeves). We assert the shadow-
free composition seam: each day's combined weight == (1/n)·the edge's extracted per-symbol weight."""
from fxhnt.application.bybit_paper_weights import sleeve_weights_and_contributions
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
events = _seed(store)
combined_by_day = combined_symbol_weights_by_day(store, unlock_events=events)
# Re-derive each sleeve's per-day per-symbol weights and confirm the combined book is exactly (1/n)·them.
per_sleeve = {sl: sleeve_weights_and_contributions(store, sl, unlock_events=events)
for sl in ("crypto_tstrend", "unlock", "xsfunding", "positioning")}
store.close()
n_sleeves = 4
checked = 0
for day, by_sleeve in combined_by_day.items():
for sleeve, sym_w in by_sleeve.items():
src = per_sleeve[sleeve].get(day, {})
for sym, w in sym_w.items():
assert math.isclose(w, (1.0 / n_sleeves) * src[sym][0], rel_tol=1e-12, abs_tol=1e-15)
checked += 1
assert checked > 0, "reconciliation asserted nothing (seed too short?)"
def test_backfill_is_idempotent_per_date() -> None:
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
events = _seed(store)
repo = PaperRepo("sqlite://", venue=BYBIT_VENUE)
repo.migrate()
backfill_bybit_paper_book(repo, store, capital=_CAPITAL, at=_AT, unlock_events=events)
first = [(p.run_date, p.equity, p.realized, p.unrealized) for p in repo.nav_history()]
pos_first = {p.run_date: [(x.sleeve, x.symbol, x.qty) for x in repo.positions_at(p.run_date)]
for p in repo.nav_history()}
# Run again — per-date replace must reproduce the SAME rows (no dups, no drift).
backfill_bybit_paper_book(repo, store, capital=_CAPITAL, at=_AT, unlock_events=events)
store.close()
second = [(p.run_date, p.equity, p.realized, p.unrealized) for p in repo.nav_history()]
pos_second = {p.run_date: [(x.sleeve, x.symbol, x.qty) for x in repo.positions_at(p.run_date)]
for p in repo.nav_history()}
assert second == first, "re-running the backfill changed the nav curve (not idempotent)"
assert pos_second == pos_first, "re-running the backfill changed the positions (not idempotent)"
def test_backfill_does_not_touch_binance_book() -> None:
import os
import tempfile
fd, path = tempfile.mkstemp(suffix=".db")
os.close(fd)
try:
dsn = f"sqlite:///{path}"
binance = PaperRepo(dsn, venue="binance")
binance.migrate()
from fxhnt.domain.paper import Position
binance.replace_positions("2026-06-24",
[Position("crypto_tstrend", "BTCUSDT", 1.0, 50_000.0, "2026-06-24")],
weight_by_sleeve={"crypto_tstrend": 1.0}, at=_AT)
before = [(p.sleeve, p.symbol, p.qty) for p in binance.latest_positions()]
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
events = _seed(store)
bybit = PaperRepo(dsn, venue=BYBIT_VENUE)
bybit.migrate()
backfill_bybit_paper_book(bybit, store, capital=_CAPITAL, at=_AT, unlock_events=events)
store.close()
assert [(p.sleeve, p.symbol, p.qty) for p in binance.latest_positions()] == before
assert binance.nav_history() == [] # bybit nav did not bleed into binance
assert len(bybit.nav_history()) > 1 # bybit multi-day book exists
finally:
os.unlink(path)
def test_backfill_incremental_resume_appends_new_days_only() -> None:
"""`since_day` (the CLI's incremental resume) fills only days strictly after it, and the resumed curve
equals a from-scratch full backfill (the per-date compute is strictly-prior, so resuming is exact)."""
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
events = _seed(store)
# Full backfill (the reference).
full_repo = PaperRepo("sqlite://", venue=BYBIT_VENUE)
full_repo.migrate()
backfill_bybit_paper_book(full_repo, store, capital=_CAPITAL, at=_AT, unlock_events=events)
full = [(p.run_date, p.equity) for p in full_repo.nav_history()]
assert len(full) > 2
# Resumed: a full first pass, then a pure resume STRICTLY AFTER the last persisted day (the CLI's
# incremental path) must be a no-op — and the resulting curve must equal the from-scratch full backfill.
inc_repo = PaperRepo("sqlite://", venue=BYBIT_VENUE)
inc_repo.migrate()
backfill_bybit_paper_book(inc_repo, store, capital=_CAPITAL, at=_AT, unlock_events=events,
since_day=None) # full first pass
last_done = inc_repo.latest_nav_run_date()
since = (dt.date.fromisoformat(last_done) - dt.date(1970, 1, 1)).days
# A pure resume from the last persisted day must be a no-op (nothing strictly after).
added = backfill_bybit_paper_book(inc_repo, store, capital=_CAPITAL, at=_AT, unlock_events=events,
since_day=since)
store.close()
assert added == 0, "resume after the last persisted day should add nothing"
inc = [(p.run_date, p.equity) for p in inc_repo.nav_history()]
assert inc == full, "incremental resume curve diverged from full backfill"
def test_backfill_is_read_only_against_the_warehouse() -> None:
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
events = _seed(store)
guarded = _WriteTripwireStore(store)
repo = PaperRepo("sqlite://", venue=BYBIT_VENUE)
repo.migrate()
n = backfill_bybit_paper_book(repo, guarded, capital=_CAPITAL, at=_AT, unlock_events=events)
store.close()
assert n > 1
assert repo.nav_history()
def test_backfill_is_memory_bounded_to_the_universe() -> None:
"""The close read goes through `_bounded_store`, so only the universe symbols load (the same memory bound
the live derive uses). With CCCUSDT excluded from the universe, `closes_by_day` never surfaces it and no
persisted position is for an out-of-universe symbol."""
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
events = _seed(store)
universe = {"BTCUSDT", "AAAUSDT", "BBBUSDT"} # deliberately EXCLUDE CCCUSDT
# The close read is universe-bounded: the excluded symbol never appears.
by_day = closes_by_day(store, universe=universe)
assert by_day
for day_closes in by_day.values():
assert set(day_closes).issubset(universe)
assert all("CCCUSDT" not in c for c in by_day.values())
repo = PaperRepo("sqlite://", venue=BYBIT_VENUE)
repo.migrate()
n = backfill_bybit_paper_book(repo, store, capital=_CAPITAL, at=_AT,
unlock_events=events, universe=universe)
store.close()
assert n > 1
# No persisted position is for an out-of-universe symbol.
for p in repo.nav_history():
for x in repo.positions_at(p.run_date):
assert x.symbol in universe
def test_closes_by_day_inverts_the_panel() -> None:
store = TimescaleFeatureStore("sqlite://", table="bybit_features")
_seed(store, days=10)
by_day = closes_by_day(store)
panel = store.crypto_close_panel()
store.close()
# Round-trip: every (symbol, day, close) in the panel appears in the inverted map.
for sym, series in panel.items():
for day, close in series.items():
assert math.isclose(by_day[int(day)][sym], float(close), rel_tol=1e-12)

92
tests/integration/test_bybit_paper_backfill_cli.py Normal file
View File

@@ -0,0 +1,92 @@
"""The `fxhnt bybit-paper-backfill` CLI: walks the full Bybit history and persists the live eq-wt 4-edge book
daily under venue="bybit" into the operational DB. NO network (in-memory bybit_features seeded into a
file-sqlite DSN shared by the warehouse store + the paper repo; the unlock loader is stubbed to the seeded
calendar). Proves the command wires args + the incremental resume, and writes a multi-day venue=bybit curve."""
from __future__ import annotations
from typer.testing import CliRunner
import fxhnt.application.bybit_liquidity as bybit_liquidity
import fxhnt.application.unlock_calendar_loader as unlock_loader
import fxhnt.cli as cli
from fxhnt.adapters.persistence.paper_repo import PaperRepo
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
from fxhnt.application.bybit_paper_book import BYBIT_VENUE
runner = CliRunner()
_DAY = 86_400
def _seed(store: TimescaleFeatureStore, *, days: int = 200) -> list[dict]:
import math as _m
syms = ["BTCUSDT", "AAAUSDT", "BBBUSDT", "CCCUSDT"]
px = {s: 100.0 for s in syms}
px["BTCUSDT"] = 50_000.0
for d in range(days):
crowd = (d % 3 == 0)
for s in syms:
drift = {"BTCUSDT": 0.001, "AAAUSDT": 0.002, "BBBUSDT": -0.001, "CCCUSDT": 0.0015}[s]
px[s] *= (1.0 + drift + 0.003 * _m.sin(d / 5.0 + hash(s) % 7))
ratio = (0.70 if crowd else 0.30) if s == "AAAUSDT" else \
(0.30 if crowd else 0.70) if s == "BBBUSDT" else 0.50
store.write_features(s, [(d * _DAY, {
"close": px[s], "funding": 0.0005, "spot_close": px[s] * 1.0002, "long_ratio": ratio})])
return [{"sym": "AAA", "cliff_day": days - 5, "n": 5_000_000.0, "cat": "insiders"}]
class _settings:
def __init__(self, dsn: str) -> None:
self.operational_dsn = dsn
self.paper_capital = 100_000.0
self.paper_enabled = True
def _wire(tmp_path, monkeypatch) -> str:
"""Seed bybit_features into a file-sqlite DB, stub settings + the unlock loader, return the DSN."""
dsn = f"sqlite:///{tmp_path / 'cockpit.db'}"
store = TimescaleFeatureStore(dsn, table="bybit_features")
events = _seed(store)
store.close()
monkeypatch.setattr(cli, "get_settings", lambda: _settings(dsn))
# Full universe (no liquid floor) so the seed's symbols all participate, and the seeded calendar.
monkeypatch.setattr(bybit_liquidity, "liquid_universe", lambda *a, **k: set())
monkeypatch.setattr(unlock_loader, "operational_unlock_repo", lambda s: None)
monkeypatch.setattr(unlock_loader, "load_unlock_events", lambda repo, path: events)
return dsn
def test_cli_persists_multi_day_bybit_curve(tmp_path, monkeypatch):
dsn = _wire(tmp_path, monkeypatch)
result = runner.invoke(cli.app, ["bybit-paper-backfill", "--min-dollar-vol", "0"])
assert result.exit_code == 0, result.output
assert "venue=bybit" in result.output
repo = PaperRepo(dsn, venue=BYBIT_VENUE)
hist = repo.nav_history()
assert len(hist) > 1, "expected a multi-day venue=bybit nav series"
dates = [p.run_date for p in hist]
assert dates == sorted(dates) == sorted(set(dates))
# The binance book is untouched (no rows under the default venue).
assert PaperRepo(dsn, venue="binance").nav_history() == []
def test_cli_incremental_resume_is_noop_then_rebuild_matches(tmp_path, monkeypatch):
dsn = _wire(tmp_path, monkeypatch)
r1 = runner.invoke(cli.app, ["bybit-paper-backfill", "--min-dollar-vol", "0"])
assert r1.exit_code == 0, r1.output
hist1 = [(p.run_date, p.equity) for p in PaperRepo(dsn, venue=BYBIT_VENUE).nav_history()]
assert len(hist1) > 1
# Default (incremental): nothing new after the last persisted day → persists 0, curve unchanged.
r2 = runner.invoke(cli.app, ["bybit-paper-backfill", "--min-dollar-vol", "0"])
assert r2.exit_code == 0, r2.output
assert "persisted 0 day(s) (incremental)" in r2.output
hist2 = [(p.run_date, p.equity) for p in PaperRepo(dsn, venue=BYBIT_VENUE).nav_history()]
assert hist2 == hist1
# --rebuild recomputes the whole history; idempotent → identical curve.
r3 = runner.invoke(cli.app, ["bybit-paper-backfill", "--min-dollar-vol", "0", "--rebuild"])
assert r3.exit_code == 0, r3.output
hist3 = [(p.run_date, p.equity) for p in PaperRepo(dsn, venue=BYBIT_VENUE).nav_history()]
assert hist3 == hist1, "rebuild diverged from the original (should be idempotent)"