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

fix(vrp): exec-twin ladder cap + per-rung sizing + combo BUY-credit sign + backfill master branch + P&L helper tests

Browse Source 
The VRP exec twin previously opened a NEW full-envelope spread every run, overwriting the
single-slot book-state and never closing the prior one -- armed weekly it would stack N x
notional. plan_and_record_vrp now maintains a 5-rung ladder (mirrors VrpStrategy.advance):
closes rungs at 50% profit-take or DTE<=1 first, then opens at most one new rung sized at
envelope/ladder (never the full envelope) only when a slot is free. Ladder decisions are
factored into a pure, unit-tested plan_vrp_ladder helper.

Also: IbkrBroker.place_combo used the wrong parent-order action for a net-credit combo (SELL
with a negative limit); switched to BUY with a negative limit per IBKR's combo convention.
fxhnt-opra-backfill.yaml's git-sync cloned the feature branch instead of master (this Job is
applied manually post-merge). Added tests/unit/test_vrp_exec_helpers.py for the previously
untested pure P&L helpers (compute_spread_exec_return/compute_ladder_exec_return/build_pos_state).
Removed the dead --live flag from execute-vrp (plan_and_record_vrp never consulted allow_live;
the paper-envelope refusal guard is the real safety mechanism).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-07-13 00:36:11 +02:00
parent 8d0c6c67f9
commit c4fb3f73eb
6 changed files with 361 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
infra/k8s/jobs/fxhnt-opra-backfill.yaml
View File

@@ -45,7 +45,7 @@ spec:
mkdir -p /root/.ssh && cp /etc/git-ssh/ssh-privatekey /root/.ssh/id_ed25519 && chmod 600 /root/.ssh/id_ed25519
printf 'Host *\n StrictHostKeyChecking no\n UserKnownHostsFile /dev/null\n' > /root/.ssh/config
rm -rf /code/src
git clone --depth 1 --branch feat/ibkr-consolidation-xsp-vrp-ucits --filter=blob:none --sparse \
git clone --depth 1 --branch master --filter=blob:none --sparse \
"ssh://git@gitea-sshd.foxhunt.svc.cluster.local:22/gitadmin/fxhnt.git" /tmp/repo
cd /tmp/repo && git sparse-checkout set src && [ -d /tmp/repo/src ]
cp -a /tmp/repo/src /code/src && echo "git-sync OK: $(git rev-parse --short HEAD)"

7
src/fxhnt/adapters/broker/ibkr.py
View File

@@ -75,7 +75,12 @@ class IbkrBroker:
combo_legs.append(ComboLeg(conId=cid, ratio=qty, action=side, exchange="SMART"))
bag = Bag(symbol="XSP", exchange="SMART", currency="USD", comboLegs=combo_legs)
total_qty = legs[0][2]
order = LimitOrder("SELL", total_qty, net_limit) # SELL the spread for a net credit
# IBKR combo convention: per-leg actions already encode direction (e.g. SELL short put / BUY long
# put to open; reversed to close). The PARENT bag order is always "BUY" the combo, with the limit
# price carrying the net cash flow sign: a NEGATIVE limit = BUY the bag "for" a negative debit, i.e.
# a net CREDIT received (opening); a POSITIVE limit = a net DEBIT paid (closing). "SELL" the bag
# would invert every leg's already-assigned action, which is wrong here.
order = LimitOrder("BUY", total_qty, net_limit)
trade = ib.placeOrder(bag, order)
ib.sleep(1)
return str(trade.order.orderId)

264
src/fxhnt/application/vrp_exec_record.py
View File

@@ -1,10 +1,21 @@
"""Executed VRP twin (sub-project C-style): place the atomic XSP put-credit-spread on IBKR and record the
executed-vs-modeled reconciliation row. SUSPENDED until the account has options permission (no live OPRA in
CI either, so the runtime path is not testable here). `build_combo_legs` and `refuse_if_live` are pure and
unit-tested; `plan_and_record_vrp` mirrors `multistrat_exec_record.plan_and_record`'s guard structure but
sizes a SINGLE defined-risk spread (not a continuous weight vector) — there is no B2a allocation for VRP yet,
so it only acts when `paper_envelope > 0`; a `paper_envelope` of 0 flattens (no B2a fallback, same convention
as `scaled_weights` flattening to cash on an empty envelope)."""
CI either, so the runtime path is not testable here).
Maintains a LADDER of open rungs in the `vrp_exec_pos` book-state (mirrors `VrpStrategy.advance`'s ladder
discipline in `vrp_book.py`, not a single overwritten position): each run marks every open rung against
today's frozen slice, closes any rung that hit 50% profit-take or DTE<=1 (placing the reverse combo), and
opens AT MOST ONE new rung — sized at `envelope / ladder`, never the full envelope — when there is a free
ladder slot on the weekly entry day. This bounds total open notional at ~envelope regardless of how many
rungs are simultaneously open; armed weekly, it can never stack N spreads at full size.
`build_combo_legs`/`build_close_legs`, `refuse_if_live`, `plan_vrp_ladder`, `rung_notional`,
`compute_spread_exec_return`/`compute_ladder_exec_return` and `build_pos_state` are pure and unit-tested.
`plan_and_record_vrp` mirrors `multistrat_exec_record.plan_and_record`'s guard structure but sizes a LADDER
of defined-risk spreads (not a continuous weight vector) — there is no B2a allocation for VRP yet, so new
entries only happen when `paper_envelope > 0`; existing rungs still mark/close on a de-funded envelope
(they cannot be silently stranded), matching the convention elsewhere that a de-funded envelope stops new
risk but does not orphan positions."""
from __future__ import annotations
import datetime as dt
@@ -13,12 +24,22 @@ from typing import Any
def build_combo_legs(*, short_osi: str, long_osi: str, short_price: float, long_price: float,
qty: int) -> tuple[list[tuple[str, str, int]], float]:
"""Defined-risk put-credit-spread: SELL the short put, BUY the wing. Net limit is negative (a credit)."""
"""Defined-risk put-credit-spread OPEN: SELL the short put, BUY the wing. Net limit is negative (a
credit) — see `IbkrBroker.place_combo` for the parent-order sign convention this feeds."""
legs = [(short_osi, "SELL", qty), (long_osi, "BUY", qty)]
net = -(short_price - long_price) # credit received -> negative limit
return legs, round(net, 2)
def build_close_legs(*, short_osi: str, long_osi: str, qty: int,
mark_today: float) -> tuple[list[tuple[str, str, int]], float]:
"""Reverse of `build_combo_legs` to FLATTEN a held rung: BUY back the short (was sold), SELL the long
(was bought). `net_limit` is POSITIVE — the debit paid to close, equal to `mark_today` (`_mark_spread`'s
short-minus-long convention)."""
legs = [(short_osi, "BUY", qty), (long_osi, "SELL", qty)]
return legs, round(float(mark_today), 2)
def refuse_if_live(paper_envelope: float, account_is_paper: bool) -> None:
"""The real-capital exposure guard, factored out pure so it is testable without a live broker/repo: a
PAPER-validation envelope (`paper_envelope > 0`) NEVER runs against a non-paper account. Must be the
@@ -39,28 +60,80 @@ def _mark_spread(bars: Any, spread: dict[str, Any], day: str) -> float | None:
return short_val - long_val
def compute_spread_exec_return(pos_prior: dict[str, Any], mark_today: float | None) -> float | None:
"""Envelope-basis executed return: the mark-to-market P&L of the spread ALREADY HELD since the last run,
as a fraction of the envelope (mirrors `multistrat_exec_record.compute_exec_return` — a newly-placed
spread's credit is a capital flow, not a gain, so it is booked on the FOLLOWING run once it has actually
moved). Short-vol: value falling = profit. The $100/contract multiplier converts the frozen per-share
option marks into real dollars. None on inception (no prior spread), a de-funded envelope, or when
today's freeze is missing a mark for the held legs (a HOLD, not a phantom zero)."""
if not pos_prior or not pos_prior.get("spread"):
return None
envelope = float(pos_prior.get("envelope", 0.0))
def rung_notional(envelope: float, ladder: int) -> float:
"""The per-rung dollar target: the envelope split evenly across the ladder's slots. Sizing every new
entry at this (instead of the full envelope) bounds total open notional at ~envelope regardless of how
many rungs are simultaneously open."""
return envelope / ladder if ladder > 0 else 0.0
def plan_vrp_ladder(open_spreads: list[dict[str, Any]], marks_today: dict[str, float | None], today: str, *,
ladder: int, profit_take: float = 0.5, entry_weekday: int = 0
) -> tuple[list[tuple[dict[str, Any], float]], bool, int]:
"""Pure ladder-discipline decision (mirrors `VrpStrategy.advance`'s close-then-open management, for the
single-day-per-run exec cadence). `marks_today` is pre-fetched by the caller (keyed by `short_osi`) so
this stays I/O-free.
Returns:
to_close — [(spread, mark)] rungs to close THIS run: hit 50% profit-take
(mark <= (1-profit_take)*credit) or DTE<=1. A rung with NO mark today (aged out of the
frozen band) is a HOLD, not a close — it cannot be closed without a price (mirrors
`compute_spread_exec_return`'s None-not-phantom-zero convention).
may_open — True iff today is the weekly entry weekday AND the ladder has a free slot once this run's
closes are applied (len(open_spreads) - len(to_close) < ladder).
open_slots — 0 or 1 (a once-daily run opens at most one new rung, mirroring `VrpStrategy.advance`)."""
to_close: list[tuple[dict[str, Any], float]] = []
for sp in open_spreads:
mark = marks_today.get(sp["short_osi"])
if mark is None:
continue
expiry = dt.date.fromisoformat(sp["expiry"])
dte = (expiry - dt.date.fromisoformat(today)).days
if mark <= (1.0 - profit_take) * sp["credit"] or dte <= 1:
to_close.append((sp, mark))
remaining = len(open_spreads) - len(to_close)
is_entry_day = dt.date.fromisoformat(today).weekday() == entry_weekday
may_open = is_entry_day and remaining < ladder
return to_close, may_open, (1 if may_open else 0)
def compute_spread_exec_return(prior_val: float, mark_today: float | None, qty: int,
envelope: float) -> float | None:
"""Envelope-basis executed return of ONE rung since its last mark: the mark-to-market P&L, as a fraction
of the shared ladder envelope. Short-vol: value falling = profit. The $100/contract multiplier converts
the frozen per-share option marks into real dollars. None on a de-funded envelope or when today's freeze
is missing a mark for the held legs (a HOLD, not a phantom zero)."""
if envelope <= 0.0 or mark_today is None:
return None
qty = int(pos_prior.get("qty", 0))
prior_val = float(pos_prior.get("last_val", 0.0))
pnl = (prior_val - mark_today) * 100.0 * qty
return pnl / envelope
def build_pos_state(spread: dict[str, Any] | None, last_val: float | None, qty: int, envelope: float,
venue: str, today: str) -> dict[str, Any]:
return {"spread": spread, "last_val": (None if last_val is None else float(last_val)),
"qty": int(qty), "envelope": float(envelope), "venue": str(venue), "last_run_date": today}
def compute_ladder_exec_return(open_spreads_prior: list[dict[str, Any]], marks_today: dict[str, float | None],
envelope_prior: float) -> float | None:
"""Sum `compute_spread_exec_return` across every rung held since the last run (mirrors
`multistrat_exec_record.compute_exec_return`'s book-level pattern, one rung at a time). None on
inception (no prior rungs), a de-funded prior envelope, or when NOT ONE rung produced a mark today (an
all-HOLD run — not a phantom zero); a rung with no mark today simply contributes 0 to the sum, same as
`VrpStrategy.advance`'s per-day loop skipping unmarked spreads."""
if not open_spreads_prior or envelope_prior <= 0.0:
return None
total = 0.0
any_marked = False
for sp in open_spreads_prior:
mark = marks_today.get(sp["short_osi"])
r = compute_spread_exec_return(float(sp.get("last_val", sp["credit"])), mark, int(sp.get("qty", 0)),
envelope_prior)
if r is not None:
total += r
any_marked = True
return total if any_marked else None
def build_pos_state(open_spreads: list[dict[str, Any]], envelope: float, venue: str,
today: str) -> dict[str, Any]:
return {"open_spreads": [dict(sp) for sp in open_spreads], "envelope": float(envelope), "venue": str(venue),
"last_run_date": today}
class _VrpExecBookingStrategy:
@@ -75,33 +148,36 @@ class _VrpExecBookingStrategy:
return [(str(extra.get("_today")), float(self._r))], extra
def _record(repo: Any, exec_return: float | None, *, spread: dict[str, Any] | None, last_val: float | None,
qty: int, envelope: float, venue: str, today: str, at: dt.datetime) -> None:
def _record(repo: Any, exec_return: float | None, *, open_spreads: list[dict[str, Any]], envelope: float,
venue: str, today: str, at: dt.datetime) -> None:
from fxhnt.application.forward_engine import run_track
run_track(repo, "vrp_exec", _VrpExecBookingStrategy(exec_return), today=today, at=at)
repo.set_book_state("vrp_exec_pos", build_pos_state(spread, last_val, qty, envelope, venue, today), at=at)
repo.set_book_state("vrp_exec_pos", build_pos_state(open_spreads, envelope, venue, today), at=at)
def plan_and_record_vrp(repo: Any, *, dbn: Any, broker: Any, nlv: float, today: str, at: dt.datetime,
paper_envelope: float, account_is_paper: bool, venue: str, execute: bool) -> dict[str, Any]:
"""Mark the PRIOR held spread (if any) against today's frozen slice, select today's candidate spread off
the SAME frozen-bars machinery `vrp_nav` uses (never touching OPRA outside the freeze call), and — when
`execute` — place it as an atomic combo and record the observed `vrp_exec` return.
"""Mark every PRIOR held rung (if any) against today's frozen slice, apply the ladder discipline (close
rungs at 50% profit / DTE<=1, cap entries at `ladder`, size a new rung at `envelope/ladder`), and — when
`execute` — place the closing/opening combos and record the observed `vrp_exec` return.
Guards (mirrors `multistrat_exec_record.plan_and_record` guard-for-guard):
1. `refuse_if_live` — paper-envelope on a non-paper account raises BEFORE any repo/broker/OPRA call
(zero orders placed, zero rows read/written).
2. per-day idempotency — a `vrp_exec_pos` book-state already stamped `today` is a no-op re-run.
3. a `broker.place_combo` failure becomes a structured `error` in the return dict (never a crash); the
prior held position is re-recorded UNCHANGED (mirrors the bybit per-order gap-row + kill-switch
discipline: a failed order must not silently advance the position basis).
3. a `broker.place_combo` failure (open OR close) becomes a structured entry in `errors` (never a
crash); a failed CLOSE re-adds the rung UNCHANGED to the ladder (mirrors the bybit per-order gap-row
+ kill-switch discipline: a failed order must not silently advance the position basis).
4. recording is skipped entirely when `execute` is False (dry-run plans only, same as multistrat)."""
refuse_if_live(paper_envelope, account_is_paper)
pos_prior = repo.get_book_state("vrp_exec_pos")
if pos_prior.get("last_run_date") == today:
return {"noop": True, "reason": f"already ran on {today}", "exec_return": None, "placed": 0, "spread": None}
return {"noop": True, "reason": f"already ran on {today}", "exec_return": None, "placed": 0,
"closed": 0, "spread": None}
open_spreads_prior: list[dict[str, Any]] = list(pos_prior.get("open_spreads", []))
envelope_prior = float(pos_prior.get("envelope", 0.0))
envelope = min(float(paper_envelope), float(nlv)) if paper_envelope > 0.0 else 0.0
from fxhnt.adapters.orchestration.assets import _freeze_xsp_slice, _xsp_forward_estimate
@@ -114,62 +190,90 @@ def plan_and_record_vrp(repo: Any, *, dbn: Any, broker: Any, nlv: float, today:
forward = _xsp_forward_estimate(dbn, today)
_freeze_xsp_slice(dbn, bars, as_of=today, forward=forward, at=at)
mark_today = _mark_spread(bars, pos_prior["spread"], today) if pos_prior.get("spread") else None
exec_return = compute_spread_exec_return(pos_prior, mark_today)
result: dict[str, Any] = {"exec_return": exec_return, "placed": 0, "spread": None, "notes": []}
if envelope <= 0.0:
result["notes"].append("no envelope (paper-envelope off, no B2a allocation for vrp_exec yet) — flat")
if execute:
_record(repo, exec_return, spread=None, last_val=None, qty=0, envelope=0.0, venue=venue,
today=today, at=at)
return result
from fxhnt.application.vrp_book import VrpStrategy
from fxhnt.registry import STRATEGY_REGISTRY
strat = VrpStrategy(bars)
spread = strat._select_spread(today) # noqa: SLF001 — composition root reaches into the sim's selector
if spread is None:
result["notes"].append("no candidate spread selected for today")
if execute:
_record(repo, exec_return, spread=None, last_val=None, qty=0, envelope=envelope, venue=venue,
today=today, at=at)
return result
ladder = int(STRATEGY_REGISTRY["vrp"]["definition"]["params"]["ladder"])
strat = VrpStrategy(bars, ladder=ladder)
marks = bars.bars_for([spread["short_osi"], spread["long_osi"]], today, today)
short_price = marks.get(spread["short_osi"], {}).get(today)
long_price = marks.get(spread["long_osi"], {}).get(today)
if short_price is None or long_price is None:
raise RuntimeError(f"vrp_exec: selected spread has no frozen mark for {today}")
marks_today = {sp["short_osi"]: _mark_spread(bars, sp, today) for sp in open_spreads_prior}
exec_return = compute_ladder_exec_return(open_spreads_prior, marks_today, envelope_prior)
qty = int(envelope // strat.margin)
if qty < 1:
result["notes"].append(f"envelope ${envelope:,.0f} < margin ${strat.margin:,.0f}/contract — no spread")
if execute:
_record(repo, exec_return, spread=None, last_val=None, qty=0, envelope=envelope, venue=venue,
today=today, at=at)
return result
to_close, may_open, _open_slots = plan_vrp_ladder(
open_spreads_prior, marks_today, today, ladder=ladder,
profit_take=strat._profit_take, entry_weekday=strat._entry_weekday) # noqa: SLF001 — composition
# root reaches into the sim's private ladder-management params so the exec twin and the backtest
# never drift apart.
legs, net = build_combo_legs(short_osi=spread["short_osi"], long_osi=spread["long_osi"],
short_price=short_price, long_price=long_price, qty=qty)
result.update(spread=spread, legs=legs, net=net, qty=qty)
close_osi = {sp["short_osi"] for sp, _mark in to_close}
kept: list[dict[str, Any]] = []
for sp in open_spreads_prior:
if sp["short_osi"] in close_osi:
continue
m = marks_today.get(sp["short_osi"])
kept.append(dict(sp, last_val=m) if m is not None else sp)
result: dict[str, Any] = {"exec_return": exec_return, "placed": 0, "closed": 0, "to_close": len(to_close),
"errors": [], "notes": [], "spread": None}
candidate: dict[str, Any] | None = None
qty_new = 0
legs_new: list[tuple[str, str, int]] | None = None
net_new: float | None = None
if envelope <= 0.0:
result["notes"].append("no envelope (paper-envelope off, no B2a allocation for vrp_exec yet) — no new rungs")
elif may_open:
candidate = strat._select_spread(today) # noqa: SLF001 — composition root reaches into the sim's selector
if candidate is None:
result["notes"].append("no candidate spread selected for today")
else:
marks = bars.bars_for([candidate["short_osi"], candidate["long_osi"]], today, today)
short_price = marks.get(candidate["short_osi"], {}).get(today)
long_price = marks.get(candidate["long_osi"], {}).get(today)
if short_price is None or long_price is None:
raise RuntimeError(f"vrp_exec: selected spread has no frozen mark for {today}")
rung_target = rung_notional(envelope, ladder)
qty_new = int(rung_target // strat.margin)
if qty_new < 1:
result["notes"].append(
f"rung target ${rung_target:,.0f} (envelope/{ladder}) < margin ${strat.margin:,.0f}/contract"
" — no new rung")
candidate = None
else:
legs_new, net_new = build_combo_legs(
short_osi=candidate["short_osi"], long_osi=candidate["long_osi"],
short_price=short_price, long_price=long_price, qty=qty_new)
result.update(spread=candidate, legs=legs_new, net=net_new, qty=qty_new)
else:
result["notes"].append("ladder full or off entry day — no new rung")
if not execute:
return result
try:
order_id = broker.place_combo(legs, net)
except Exception as e: # a broker hiccup becomes a visible gap, never a crash — re-record the UNCHANGED
# prior position so a failed order never silently advances the return basis.
result["error"] = f"{type(e).__name__}: {str(e)[:120]}"
_record(repo, exec_return, spread=pos_prior.get("spread"), last_val=mark_today,
qty=int(pos_prior.get("qty", 0)), envelope=envelope, venue=venue, today=today, at=at)
return result
open_spreads_next = list(kept)
result["placed"] = 1
result["order_id"] = order_id
entry_val = short_price - long_price
_record(repo, exec_return, spread=spread, last_val=entry_val, qty=qty, envelope=envelope, venue=venue,
today=today, at=at)
for sp, mark in to_close:
close_legs, close_net = build_close_legs(short_osi=sp["short_osi"], long_osi=sp["long_osi"],
qty=int(sp["qty"]), mark_today=mark)
try:
broker.place_combo(close_legs, close_net)
result["closed"] += 1
except Exception as e: # a failed CLOSE must not silently drop the rung -- re-add it UNCHANGED.
result["errors"].append(f"close {sp['short_osi']}: {type(e).__name__}: {str(e)[:120]}")
open_spreads_next.append(sp)
if candidate is not None and legs_new is not None and net_new is not None:
try:
order_id = broker.place_combo(legs_new, net_new)
result["placed"] = 1
result["order_id"] = order_id
opened = dict(candidate, qty=qty_new, last_val=candidate["credit"])
open_spreads_next.append(opened)
result["spread"] = opened
except Exception as e: # a failed OPEN just means no new rung this run -- ladder stays as-is.
result["errors"].append(f"open: {type(e).__name__}: {str(e)[:120]}")
result["spread"] = None
_record(repo, exec_return, open_spreads=open_spreads_next, envelope=envelope, venue=venue, today=today, at=at)
result["open_spreads"] = len(open_spreads_next)
return result

19
src/fxhnt/cli.py
View File

@@ -362,14 +362,15 @@ def execute_bybit(
@app.command("execute-vrp")
def execute_vrp(
do_execute: bool = typer.Option(False, "--execute", help="place the combo order (default: dry-run plan only)"),
live: bool = typer.Option(False, "--live", help="allow a non-paper account"),
do_execute: bool = typer.Option(False, "--execute", help="place combo orders (default: dry-run plan only)"),
paper_envelope: float = typer.Option(
0.0, "--paper-envelope",
help="PAPER-only validation envelope in $ (there is no B2a allocation for VRP yet, so this is the "
"ONLY way this track sizes a spread; 0 = no action)."),
"ONLY way this track sizes a new ladder rung; 0 = no new rungs, existing rungs still mark/close)."),
) -> None:
"""Reconcile + record the EXECUTED XSP put-credit-spread VRP twin on IBKR (`mleg` combo). SUSPENDED: the
"""Reconcile + record the EXECUTED XSP put-credit-spread VRP twin on IBKR (`mleg` combo), maintaining a
ladder of open rungs (mirrors `VrpStrategy.advance`'s discipline: close a rung at 50% profit or DTE<=1,
cap entries at `ladder`, size each new rung at envelope/ladder — never the full envelope). SUSPENDED: the
IBKR account currently lacks options trading permission and CI has no live OPRA, so this path is coded
but not runnable here — dry-run by default even once armed. Same paper-envelope refusal guard as
`execute-multistrat`/`execute-bybit`. Records the observed `vrp_exec` return."""
@@ -379,8 +380,6 @@ def execute_vrp(
from fxhnt.application.vrp_exec_record import plan_and_record_vrp
settings = get_settings()
if live:
settings.execution.allow_live = True
d = settings.databento
dbn = DatabentoDataProvider(max_cost_usd=d.max_cost_usd, equity_dataset=d.equity_dataset,
future_dataset=d.future_dataset, default_start=d.default_start)
@@ -411,15 +410,17 @@ def execute_vrp(
if out.get("noop"):
typer.echo(f"-> no-op: {out.get('reason')}")
return
typer.echo(f"to_close={out.get('to_close', 0)} closed={out.get('closed', 0)}")
spread = out.get("spread")
if spread:
typer.echo(f"spread: SELL {spread['short_osi']} / BUY {spread['long_osi']} qty={out.get('qty')} "
f"net ${out.get('net'):+.2f}")
for n in out.get("notes", []):
typer.echo(f" - {n}")
if out.get("error"):
typer.echo(f" order error: {out['error']}")
typer.echo(f"-> exec_return {out.get('exec_return')} | placed {out.get('placed', 0)}")
for err in out.get("errors", []):
typer.echo(f" order error: {err}")
typer.echo(f"-> exec_return {out.get('exec_return')} | placed {out.get('placed', 0)} "
f"| open_spreads {out.get('open_spreads')}")
@app.command("migrate-cockpit")

80
tests/unit/test_vrp_exec_helpers.py Normal file
View File

@@ -0,0 +1,80 @@
import pytest
from fxhnt.application.vrp_exec_record import (
build_pos_state,
compute_ladder_exec_return,
compute_spread_exec_return,
)
def test_compute_spread_exec_return_envelope_zero_is_none():
assert compute_spread_exec_return(prior_val=1.0, mark_today=0.8, qty=2, envelope=0.0) is None
def test_compute_spread_exec_return_no_mark_today_is_hold_not_phantom_zero():
assert compute_spread_exec_return(prior_val=1.0, mark_today=None, qty=2, envelope=200_000.0) is None
def test_compute_spread_exec_return_winning_spread_is_positive_credit_decay():
# short-vol: credit decayed from 1.00 -> 0.60 = profit. qty=2, $100/contract multiplier.
r = compute_spread_exec_return(prior_val=1.00, mark_today=0.60, qty=2, envelope=200_000.0)
assert r == pytest.approx((1.00 - 0.60) * 100.0 * 2 / 200_000.0)
assert r > 0.0
def test_compute_spread_exec_return_losing_spread_is_negative_and_bounded_by_width():
# a defined-risk put-credit-spread's worst mark is bounded by the wing width ($5 here = $500/contract);
# the value rising toward that cap is a loss, correctly signed negative.
r = compute_spread_exec_return(prior_val=1.00, mark_today=4.50, qty=2, envelope=200_000.0)
assert r == pytest.approx((1.00 - 4.50) * 100.0 * 2 / 200_000.0)
assert r < 0.0
def test_compute_ladder_exec_return_inception_no_prior_rungs_is_none():
assert compute_ladder_exec_return([], {}, 200_000.0) is None
def test_compute_ladder_exec_return_definanced_envelope_is_none():
prior = [{"short_osi": "S1", "long_osi": "L1", "credit": 1.0, "last_val": 1.0, "qty": 2}]
assert compute_ladder_exec_return(prior, {"S1": 0.5}, 0.0) is None
def test_compute_ladder_exec_return_sums_only_marked_rungs():
prior = [
{"short_osi": "S1", "long_osi": "L1", "credit": 1.0, "last_val": 1.0, "qty": 2},
{"short_osi": "S2", "long_osi": "L2", "credit": 1.0, "last_val": 1.0, "qty": 2},
]
marks = {"S1": 0.60, "S2": None} # S2 has no mark today -> HOLD, contributes 0 (not a phantom zero)
r = compute_ladder_exec_return(prior, marks, 200_000.0)
assert r == pytest.approx((1.00 - 0.60) * 100.0 * 2 / 200_000.0) # only S1's marked pnl
def test_compute_ladder_exec_return_all_holds_is_none():
prior = [{"short_osi": "S1", "long_osi": "L1", "credit": 1.0, "last_val": 1.0, "qty": 2}]
assert compute_ladder_exec_return(prior, {"S1": None}, 200_000.0) is None
def test_build_pos_state_round_trip():
opens = [{"entry": "2026-07-06", "expiry": "2026-08-15", "short_osi": "S1", "long_osi": "L1",
"credit": 1.0, "last_val": 0.6, "qty": 2}]
st = build_pos_state(opens, envelope=200_000.0, venue="ibkr-paper", today="2026-07-13")
assert st["open_spreads"] == opens
assert st["envelope"] == 200_000.0
assert st["venue"] == "ibkr-paper"
assert st["last_run_date"] == "2026-07-13"
def test_build_pos_state_defensively_copies_rungs():
opens = [{"entry": "2026-07-06", "expiry": "2026-08-15", "short_osi": "S1", "long_osi": "L1",
"credit": 1.0, "last_val": 0.6, "qty": 2}]
st = build_pos_state(opens, envelope=200_000.0, venue="ibkr-paper", today="2026-07-13")
opens.append({"short_osi": "S2"}) # mutate the caller's list after the call
opens[0]["last_val"] = 999.0 # mutate a rung dict in the caller's list
assert len(st["open_spreads"]) == 1
assert st["open_spreads"][0]["last_val"] == 0.6
def test_build_pos_state_empty_ladder():
st = build_pos_state([], envelope=0.0, venue="", today="2026-07-13")
assert st["open_spreads"] == []
assert st["envelope"] == 0.0

81
tests/unit/test_vrp_exec_record.py
View File

@@ -1,6 +1,12 @@
import pytest
from fxhnt.application.vrp_exec_record import build_combo_legs, refuse_if_live
from fxhnt.application.vrp_exec_record import (
build_close_legs,
build_combo_legs,
plan_vrp_ladder,
refuse_if_live,
rung_notional,
)
def test_build_combo_legs_is_defined_risk_credit():
@@ -11,6 +17,79 @@ def test_build_combo_legs_is_defined_risk_credit():
assert net == pytest.approx(-0.80) # negative limit = net credit received
def test_build_close_legs_reverses_open_legs_with_positive_debit():
legs, net = build_close_legs(short_osi="XSP..P00470000", long_osi="XSP..P00465000", qty=1, mark_today=0.30)
assert ("XSP..P00470000", "BUY", 1) in legs # BUY back the short
assert ("XSP..P00465000", "SELL", 1) in legs # SELL the long
assert net == pytest.approx(0.30) # positive limit = debit paid to close
def _rung(*, short_osi="S1", long_osi="L1", credit=1.0, qty=2, expiry="2026-09-01"):
return {"entry": "2026-07-06", "expiry": expiry, "short_osi": short_osi, "long_osi": long_osi,
"credit": credit, "last_val": credit, "qty": qty}
def test_plan_vrp_ladder_no_stacking_when_ladder_is_full():
# 5 rungs already open, none hitting profit-take/DTE -> no closes -> ladder stays full -> may NOT open.
opens = [_rung(short_osi=f"S{i}") for i in range(5)]
marks = {sp["short_osi"]: 0.9 for sp in opens} # well above the 50% profit-take line, DTE far out
to_close, may_open, slots = plan_vrp_ladder(opens, marks, "2026-07-06", ladder=5, entry_weekday=0)
assert to_close == []
assert may_open is False
assert slots == 0
def test_plan_vrp_ladder_closes_rung_at_dte_le_1():
sp = _rung(expiry="2026-07-07") # DTE = 1 on "2026-07-06"
to_close, _may_open, _slots = plan_vrp_ladder([sp], {"S1": 0.9}, "2026-07-06", ladder=5, entry_weekday=0)
assert len(to_close) == 1
assert to_close[0][0] is sp
def test_plan_vrp_ladder_closes_rung_at_profit_take():
sp = _rung(credit=1.0)
# mark <= (1 - profit_take) * credit == 0.50 -> closes
to_close, _may_open, _slots = plan_vrp_ladder([sp], {"S1": 0.40}, "2026-07-06", ladder=5,
profit_take=0.5, entry_weekday=0)
assert len(to_close) == 1
def test_plan_vrp_ladder_holds_rung_with_no_mark_today():
sp = _rung()
to_close, _may_open, _slots = plan_vrp_ladder([sp], {"S1": None}, "2026-07-06", ladder=5, entry_weekday=0)
assert to_close == [] # no mark -> HOLD, not a close
def test_plan_vrp_ladder_may_open_on_entry_day_with_free_slot():
to_close, may_open, slots = plan_vrp_ladder([], {}, "2026-07-06", ladder=5, entry_weekday=0) # a Monday
assert to_close == []
assert may_open is True
assert slots == 1
def test_plan_vrp_ladder_no_open_off_entry_day():
to_close, may_open, slots = plan_vrp_ladder([], {}, "2026-07-07", ladder=5, entry_weekday=0) # a Tuesday
assert may_open is False
assert slots == 0
def test_plan_vrp_ladder_open_frees_up_when_a_close_makes_room():
# ladder full (5/5), but one rung closes this run -> a slot opens for a new entry.
opens = [_rung(short_osi=f"S{i}", expiry="2026-07-07") if i == 0 else _rung(short_osi=f"S{i}")
for i in range(5)]
marks = {sp["short_osi"]: 0.9 for sp in opens}
to_close, may_open, slots = plan_vrp_ladder(opens, marks, "2026-07-06", ladder=5, entry_weekday=0)
assert len(to_close) == 1
assert may_open is True
assert slots == 1
def test_rung_notional_splits_envelope_across_ladder_and_bounds_total():
per_rung = rung_notional(1_000_000.0, 5)
assert per_rung == pytest.approx(200_000.0)
assert per_rung * 5 == pytest.approx(1_000_000.0) # total open notional bounded at ~envelope
class _RaisingBroker:
"""A fake broker whose place_combo must NEVER be called by a refused (non-paper, real-capital) run."""
def __init__(self) -> None: