Haozhe Wu
d3c76819c3
Fix live option guards, grounding operand evidence, source-bound benchmarks, trade artifacts, attribution, drawdowns, valuation tolerances, and SEC point-in-time factor inputs. Add Brazil research data and market support, 13 fundamental and valuation factors, channel progress forwarding, and opt-in ZimaOS templates. Update seven README News sections with three visible dates. Validation: 17,635 backend tests passed; 1,084 final targeted tests and 103 frontend tests passed. Frontend build, package build, documentation checks, and CI gates passed.
1112 lines
44 KiB
Python
1112 lines
44 KiB
Python
"""Tests for BaseEngine shared logic: _align, _close_position, _calc_equity.
|
||
|
||
Uses ChinaAEngine as a concrete implementation since BaseEngine is abstract.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from dataclasses import FrozenInstanceError, replace
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import pytest
|
||
|
||
from backtest.engines.base import BaseEngine, _align, _load_optimizer
|
||
from backtest.engines.china_a import ChinaAEngine
|
||
from backtest.models import Position
|
||
|
||
_ENTRY_TS = pd.Timestamp("2026-01-02")
|
||
_ADJUST_TS = pd.Timestamp("2026-01-03")
|
||
|
||
|
||
class _AdjustmentEngine(BaseEngine):
|
||
def __init__(self, **overrides):
|
||
config = {"initial_cash": 1_000.0, "leverage": 1.0, "position_adjustment": "rebalance"}
|
||
config.update(overrides)
|
||
super().__init__(config)
|
||
self.bar_positions: list[dict[str, Position]] = []
|
||
self.bar_capitals: list[float] = []
|
||
self.adjustment_events: list[dict] = []
|
||
|
||
def can_execute(self, symbol, direction, bar):
|
||
return True
|
||
|
||
def round_size(self, raw_size, price):
|
||
return round(max(raw_size, 0.0), 6)
|
||
|
||
def calc_commission(self, size, price, direction, is_open):
|
||
return size * price * float(self.config.get("fee_rate", 0.0))
|
||
|
||
def apply_slippage(self, price, direction):
|
||
return price * (1 + direction * float(self.config.get("slippage", 0.0)))
|
||
|
||
def after_position_adjustment(self, **event):
|
||
self.adjustment_events.append(event)
|
||
|
||
def after_rebalance_bar(self, timestamp, data_map, codes):
|
||
self.bar_positions.append(dict(self.positions))
|
||
self.bar_capitals.append(self.capital)
|
||
return False
|
||
|
||
|
||
class _ChangingLeverageAdjustmentEngine(_AdjustmentEngine):
|
||
def _leverage_for_symbol(self, symbol):
|
||
return 1.0 if self._bar_idx == 0 else 2.0
|
||
|
||
|
||
class _SymbolRulesAdjustmentEngine(_AdjustmentEngine):
|
||
def round_size(self, raw_size, price):
|
||
lot = {"A": 1.0, "B": 0.25}[self._active_symbol]
|
||
return int(max(raw_size, 0.0) / lot) * lot
|
||
|
||
def calc_commission(self, size, price, direction, is_open):
|
||
return size * price * {"A": 0.01, "B": 0.02}[self._active_symbol]
|
||
|
||
|
||
class _ForcedFillAdjustmentEngine(_AdjustmentEngine):
|
||
def apply_slippage(self, price, direction):
|
||
forced = self.config.get("forced_fill_price")
|
||
return super().apply_slippage(price, direction) if forced is None else forced
|
||
|
||
|
||
def _run_adjustments(
|
||
engine: _AdjustmentEngine,
|
||
weights: dict[str, list[float]],
|
||
*,
|
||
execution_prices: dict[str, list[float]] | None = None,
|
||
codes: list[str] | None = None,
|
||
) -> None:
|
||
dates = pd.date_range("2026-01-02", periods=len(next(iter(weights.values()))))
|
||
prices = {symbol: (execution_prices or {}).get(symbol, [100.0] * len(dates)) for symbol in weights}
|
||
data_map = {symbol: pd.DataFrame({"open": values, "close": values}, index=dates) for symbol, values in prices.items()}
|
||
engine._execute_bars(
|
||
dates,
|
||
data_map,
|
||
pd.DataFrame(prices, index=dates),
|
||
pd.DataFrame(weights, index=dates),
|
||
codes or list(weights),
|
||
)
|
||
|
||
|
||
def _position(direction=1, size=5.0):
|
||
return Position("A", direction, 100.0, _ENTRY_TS, size)
|
||
|
||
|
||
def _rebalance_once(engine, target_weight, raw_price=100.0):
|
||
frame = pd.DataFrame({"open": [raw_price], "close": [100.0]}, index=[_ADJUST_TS])
|
||
engine._execute_target_rebalance({"A": target_weight}, {"A": frame}, _ADJUST_TS, 1_000.0, ["A"])
|
||
|
||
|
||
def _assert_unchanged(engine, positions=None):
|
||
assert engine.capital == 1_000.0
|
||
assert engine.positions == ({} if positions is None else positions)
|
||
assert engine.trades == []
|
||
assert engine.adjustment_events == []
|
||
|
||
|
||
def _run_both_code_orders(engine_type, weights):
|
||
first, second = engine_type(), engine_type()
|
||
_run_adjustments(first, weights, codes=["A", "B"])
|
||
_run_adjustments(second, weights, codes=["B", "A"])
|
||
return first, second
|
||
|
||
|
||
def _sizes(state):
|
||
return {symbol: position.size for symbol, position in state.items()}
|
||
|
||
|
||
def test_position_adjustment_rejects_unknown_mode():
|
||
with pytest.raises(ValueError, match="position_adjustment"):
|
||
_AdjustmentEngine(position_adjustment="resize")
|
||
|
||
|
||
def test_hold_mode_keeps_same_direction_size():
|
||
engine = _AdjustmentEngine(position_adjustment="hold")
|
||
_run_adjustments(engine, {"A": [0.25, 0.50, 0.20]})
|
||
assert [state["A"].size for state in engine.bar_positions] == [2.5, 2.5, 2.5]
|
||
|
||
|
||
def test_hold_mode_preserves_legacy_negative_open_support():
|
||
engine = _AdjustmentEngine(position_adjustment="hold", allow_nonpositive_prices=True)
|
||
_run_adjustments(engine, {"A": [0.50]}, execution_prices={"A": [-100.0]})
|
||
assert engine.bar_positions[0]["A"].entry_price == -100.0
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("existing", "target_weight"), [(False, 0.50), (True, 0.0), (True, -0.50), (True, 0.80), (True, 0.20)],
|
||
ids=("initial-open", "full-close", "reversal", "increase", "reduction"))
|
||
@pytest.mark.parametrize(
|
||
("raw_price", "forced_fill"),
|
||
[(0.0, None), (-1.0, None), (np.nan, None), (np.inf, None), (100.0, 0.0), (100.0, -1.0), (100.0, np.nan), (100.0, np.inf)],
|
||
ids=("zero-raw", "negative-raw", "nan-raw", "infinite-raw", "zero-fill", "negative-fill", "nan-fill", "infinite-fill"),
|
||
)
|
||
def test_rebalance_rejects_invalid_execution_prices_before_mutation(
|
||
existing, target_weight, raw_price, forced_fill
|
||
):
|
||
engine = _ForcedFillAdjustmentEngine(
|
||
allow_nonpositive_prices=True, forced_fill_price=forced_fill
|
||
)
|
||
if existing:
|
||
engine.positions["A"] = _position()
|
||
positions = dict(engine.positions)
|
||
|
||
with pytest.raises(ValueError, match="positive execution price"):
|
||
_rebalance_once(engine, target_weight, raw_price)
|
||
_assert_unchanged(engine, positions)
|
||
|
||
|
||
def test_rebalance_empty_zero_target_ignores_invalid_price():
|
||
engine = _ForcedFillAdjustmentEngine(
|
||
allow_nonpositive_prices=True, forced_fill_price=0.0
|
||
)
|
||
_rebalance_once(engine, 0.0, raw_price=0.0)
|
||
_assert_unchanged(engine)
|
||
|
||
|
||
def test_rebalance_increases_then_reduces_same_direction_position():
|
||
engine = _AdjustmentEngine()
|
||
_run_adjustments(engine, {"A": [0.25, 0.50, 0.20]})
|
||
assert [state["A"].size for state in engine.bar_positions] == [2.5, 5.0, 2.0]
|
||
partial = next(t for t in engine.trades if t.exit_reason == "target_rebalance")
|
||
assert partial.size == 3.0
|
||
assert partial.entry_margin == 300.0
|
||
assert partial.pnl == 0.0
|
||
|
||
|
||
def test_rebalance_persists_immutable_fill_deltas_and_weighted_holding():
|
||
engine = _AdjustmentEngine()
|
||
_run_adjustments(engine, {"A": [0.25, 0.50, 0.20]})
|
||
|
||
assert [fill.action for fill in engine.fill_records] == [
|
||
"open",
|
||
"increase",
|
||
"reduce",
|
||
"close",
|
||
]
|
||
assert [fill.signed_quantity for fill in engine.fill_records] == pytest.approx(
|
||
[2.5, 2.5, -3.0, -2.0]
|
||
)
|
||
assert [fill.notional for fill in engine.fill_records] == pytest.approx(
|
||
[250.0, 250.0, 300.0, 200.0]
|
||
)
|
||
assert [fill.margin for fill in engine.fill_records] == pytest.approx(
|
||
[250.0, 250.0, 300.0, 200.0]
|
||
)
|
||
assert [fill.execution_price for fill in engine.fill_records] == pytest.approx(
|
||
[100.0, 100.0, 100.0, 100.0]
|
||
)
|
||
assert [fill.fee for fill in engine.fill_records] == pytest.approx([0.0] * 4)
|
||
assert [fill.holding_bars for fill in engine.fill_records] == [None, None, 1.5, 1.5]
|
||
assert [trade.holding_bars for trade in engine.trades] == pytest.approx([1.5, 1.5])
|
||
|
||
with pytest.raises(FrozenInstanceError):
|
||
engine.fill_records[0].fee = 1.0 # type: ignore[misc]
|
||
|
||
|
||
def test_fill_delta_artifact_is_jsonl_and_exposes_weighted_holding(tmp_path):
|
||
engine = _AdjustmentEngine()
|
||
weights = {"A": [0.25, 0.50, 0.20]}
|
||
_run_adjustments(engine, weights)
|
||
dates = pd.date_range("2026-01-02", periods=3)
|
||
prices = pd.DataFrame({"open": 100.0, "close": 100.0}, index=dates)
|
||
equity = pd.Series(
|
||
[snapshot.equity for snapshot in engine.equity_snapshots], index=dates
|
||
)
|
||
engine._write_artifacts(
|
||
tmp_path,
|
||
{"A": prices},
|
||
dates,
|
||
equity,
|
||
pd.Series(1_000.0, index=dates),
|
||
pd.Series(0.0, index=dates),
|
||
pd.DataFrame(weights, index=dates),
|
||
{},
|
||
["A"],
|
||
)
|
||
|
||
rows = [
|
||
json.loads(line)
|
||
for line in (tmp_path / "artifacts" / "fills.jsonl").read_text(
|
||
encoding="utf-8"
|
||
).splitlines()
|
||
]
|
||
assert [row["action"] for row in rows] == ["open", "increase", "reduce", "close"]
|
||
assert rows[2]["holding_bars"] == pytest.approx(1.5)
|
||
|
||
trades = pd.read_csv(tmp_path / "artifacts" / "trades.csv")
|
||
exits = trades[trades["pnl"].notna() & (trades["reason"] != "signal")]
|
||
assert exits["holding_bars"].tolist() == pytest.approx([1.5, 1.5])
|
||
|
||
|
||
def test_trades_csv_entry_row_is_the_opening_fill_after_a_scale_in(tmp_path):
|
||
"""The entry row must be a trade that happened, not the running average cost.
|
||
|
||
A scale-in moves the position's weighted entry price, but the entry row is
|
||
stamped with the first open's date, so pairing the two would describe a
|
||
fill that never took place. Each round trip still writes exactly two rows.
|
||
"""
|
||
from src.strategy_discovery.run_artifacts import read_trade_activity
|
||
|
||
engine = _AdjustmentEngine()
|
||
weights = {"A": [0.25, 0.50, 0.0, 0.25, 0.0]}
|
||
execution_prices = {"A": [100.0, 120.0, 130.0, 110.0, 115.0]}
|
||
_run_adjustments(engine, weights, execution_prices=execution_prices)
|
||
dates = pd.date_range("2026-01-02", periods=5)
|
||
prices = pd.DataFrame(
|
||
{"open": execution_prices["A"], "close": execution_prices["A"]}, index=dates
|
||
)
|
||
equity = pd.Series(
|
||
[snapshot.equity for snapshot in engine.equity_snapshots], index=dates
|
||
)
|
||
engine._write_artifacts(
|
||
tmp_path,
|
||
{"A": prices},
|
||
dates,
|
||
equity,
|
||
pd.Series(1_000.0, index=dates),
|
||
pd.Series(0.0, index=dates),
|
||
pd.DataFrame(weights, index=dates),
|
||
{},
|
||
["A"],
|
||
)
|
||
|
||
assert [fill.action for fill in engine.fill_records] == [
|
||
"open",
|
||
"increase",
|
||
"close",
|
||
"open",
|
||
"close",
|
||
]
|
||
first_open, _, _, second_open, _ = engine.fill_records
|
||
assert engine.trades[0].entry_price != pytest.approx(first_open.execution_price)
|
||
|
||
path = tmp_path / "artifacts" / "trades.csv"
|
||
trades = pd.read_csv(path)
|
||
assert len(trades) == 2 * len(engine.trades) == 4
|
||
entries = trades[trades["event"] == "entry"]
|
||
assert entries["timestamp"].tolist() == [
|
||
str(first_open.timestamp.date()),
|
||
str(second_open.timestamp.date()),
|
||
]
|
||
assert entries["price"].tolist() == pytest.approx(
|
||
[first_open.execution_price, second_open.execution_price]
|
||
)
|
||
assert entries["qty"].tolist() == pytest.approx(
|
||
[abs(first_open.signed_quantity), abs(second_open.signed_quantity)]
|
||
)
|
||
assert entries["reason"].tolist() == ["signal", "signal"]
|
||
|
||
exit_dates, max_concurrent = read_trade_activity(path)
|
||
assert [str(day) for day in exit_dates] == [
|
||
str(trade.exit_time.date()) for trade in engine.trades
|
||
]
|
||
assert max_concurrent == 1
|
||
|
||
|
||
def test_trades_csv_entry_row_falls_back_to_the_trade_without_an_opening_fill(
|
||
tmp_path,
|
||
):
|
||
"""A position that carries no opening record keeps the trade's own fields."""
|
||
engine = _AdjustmentEngine()
|
||
engine.positions["A"] = _position(size=5.0)
|
||
engine._close_position("A", 110.0, _ADJUST_TS, "signal")
|
||
dates = pd.DatetimeIndex([_ENTRY_TS, _ADJUST_TS])
|
||
engine._write_artifacts(
|
||
tmp_path,
|
||
{},
|
||
dates,
|
||
pd.Series(1_000.0, index=dates),
|
||
pd.Series(1_000.0, index=dates),
|
||
pd.Series(0.0, index=dates),
|
||
pd.DataFrame({"A": [1.0, 0.0]}, index=dates),
|
||
{},
|
||
["A"],
|
||
)
|
||
|
||
trades = pd.read_csv(tmp_path / "artifacts" / "trades.csv")
|
||
assert len(trades) == 2
|
||
entry = trades.iloc[0]
|
||
assert (entry["timestamp"], entry["price"], entry["qty"]) == (
|
||
str(_ENTRY_TS.date()),
|
||
100.0,
|
||
5.0,
|
||
)
|
||
|
||
|
||
def test_partial_reduction_allocates_nonzero_entry_and_exit_fees():
|
||
engine = _AdjustmentEngine(fee_rate=0.01)
|
||
_run_adjustments(engine, {"A": [0.50, 0.20]})
|
||
|
||
partial = next(t for t in engine.trades if t.exit_reason == "target_rebalance")
|
||
event = next(e for e in engine.adjustment_events if e["action"] == "partial_reduction")
|
||
remaining = engine.bar_positions[1]["A"]
|
||
allocated_entry_fee = event["before"].entry_commission - remaining.entry_commission
|
||
assert (allocated_entry_fee, event["trading_fee"]) == pytest.approx((3.01, 3.01))
|
||
assert (partial.commission, remaining.entry_commission) == pytest.approx((6.02, 1.99))
|
||
assert engine.bar_capitals[1] == pytest.approx(792.99)
|
||
|
||
|
||
def test_rebalance_scale_in_uses_weighted_average_entry():
|
||
engine = _AdjustmentEngine()
|
||
_run_adjustments(
|
||
engine,
|
||
{"A": [0.25, 0.50]},
|
||
execution_prices={"A": [100.0, 120.0]},
|
||
)
|
||
assert engine.bar_positions[0]["A"].size == 2.5
|
||
assert engine.bar_positions[1]["A"].size == 4.375
|
||
assert engine.bar_positions[1]["A"].entry_price == pytest.approx(108.5714285714)
|
||
|
||
|
||
def test_rebalance_rejects_same_direction_adjustment_with_changed_leverage():
|
||
engine = _ChangingLeverageAdjustmentEngine()
|
||
with pytest.raises(ValueError, match="leverage"):
|
||
_run_adjustments(engine, {"A": [0.25, 0.50]})
|
||
assert engine.capital == 750.0
|
||
assert engine.bar_positions == [
|
||
{"A": replace(_position(size=2.5), open_price=100.0, open_size=2.5)}
|
||
]
|
||
|
||
|
||
def test_rebalance_reduces_short_with_correct_signed_pnl():
|
||
engine = _AdjustmentEngine()
|
||
_run_adjustments(
|
||
engine,
|
||
{"A": [-0.50, -0.20]},
|
||
execution_prices={"A": [100.0, 90.0]},
|
||
)
|
||
partial = next(t for t in engine.trades if t.exit_reason == "target_rebalance")
|
||
assert partial.direction == -1
|
||
assert partial.size == pytest.approx(2.666667)
|
||
assert partial.pnl == pytest.approx(26.66667, rel=1e-5)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("direction", "target_weight", "expected_size", "expected_price", "action"),
|
||
[(1, 0.80, 7.272727, 110.0, "increase"), (-1, -0.80, 8.888889, 90.0, "increase"), (1, 0.20, 2.222222, 90.0, "partial_reduction"), (-1, -0.20, 1.818182, 110.0, "partial_reduction")])
|
||
def test_existing_rebalance_sizes_target_at_action_fill(
|
||
direction, target_weight, expected_size, expected_price, action
|
||
):
|
||
engine = _AdjustmentEngine(slippage=0.10)
|
||
engine.positions["A"] = _position(direction)
|
||
_rebalance_once(engine, target_weight)
|
||
|
||
assert engine.positions["A"].size == expected_size
|
||
assert engine.adjustment_events[-1]["action"] == action
|
||
assert engine.adjustment_events[-1]["execution_price"] == pytest.approx(expected_price)
|
||
|
||
|
||
@pytest.mark.parametrize("direction", [1, -1])
|
||
def test_existing_rebalance_does_not_churn_inside_slippage_band(direction):
|
||
engine = _AdjustmentEngine(slippage=0.10)
|
||
before = _position(direction)
|
||
engine.positions["A"] = before
|
||
_rebalance_once(engine, direction * 0.50)
|
||
_assert_unchanged(engine, {"A": before})
|
||
|
||
|
||
def test_rebalance_overcommitted_single_basket_scales_instead_of_aborting():
|
||
"""#1274: 100% target plus commission scales the basket, not abort."""
|
||
engine = _AdjustmentEngine(fee_rate=0.10)
|
||
_run_adjustments(engine, {"A": [1.0]})
|
||
|
||
position = engine.bar_positions[0]["A"]
|
||
assert position.size == pytest.approx(10.0 * 1000.0 / 1100.0, rel=1e-3)
|
||
assert engine.bar_capitals[0] >= 0.0
|
||
|
||
|
||
def test_rebalance_basket_with_commission_scales_to_fit():
|
||
"""#1274: a 100% target basket plus fees fills proportionally scaled."""
|
||
engine = _AdjustmentEngine(fee_rate=0.001)
|
||
_run_adjustments(engine, {"A": [0.60], "B": [0.40]})
|
||
|
||
sizes = _sizes(engine.bar_positions[0])
|
||
# One common scale factor: each sleeve keeps its share of the portfolio.
|
||
assert sizes["A"] == pytest.approx(6.0 * 1000.0 / 1001.0, rel=1e-3)
|
||
assert sizes["A"] / sizes["B"] == pytest.approx(0.60 / 0.40, rel=1e-4)
|
||
assert 0.0 <= engine.bar_capitals[0] < 0.01
|
||
|
||
|
||
def test_rebalance_scaled_basket_is_independent_of_input_code_order():
|
||
"""#1274: scaling preserves the fairness contract of the open path."""
|
||
|
||
class _FeeAdjustmentEngine(_AdjustmentEngine):
|
||
def __init__(self):
|
||
super().__init__(fee_rate=0.001)
|
||
|
||
first, second = _run_both_code_orders(_FeeAdjustmentEngine, {"A": [0.60], "B": [0.40]})
|
||
assert first.bar_positions == second.bar_positions
|
||
assert first.bar_capitals == second.bar_capitals
|
||
|
||
|
||
def test_rebalance_scaled_sizes_keep_weights_with_differing_fees():
|
||
"""#1274: one common size factor — per-symbol fees must not re-weight.
|
||
|
||
A cost-weighted per-order scale would keep the 60/40 *spend* split while
|
||
distorting sizes. Sizes must stay near the 1.5 ratio, and each symbol
|
||
must be rounded and commissioned under its own rules while scaling —
|
||
a stale active symbol would charge B's 2% fee on A's fill.
|
||
"""
|
||
|
||
class _SymbolFeeAdjustmentEngine(_AdjustmentEngine):
|
||
def round_size(self, raw_size, price):
|
||
lot = {"A": 0.05, "B": 0.05}[self._active_symbol]
|
||
return round(max(raw_size, 0.0) / lot) * lot
|
||
|
||
def calc_commission(self, size, price, direction, is_open):
|
||
rate = {"A": 0.01, "B": 0.02}[self._active_symbol]
|
||
return size * price * rate
|
||
|
||
engine = _SymbolFeeAdjustmentEngine(fee_rate=0.0)
|
||
_run_adjustments(
|
||
engine,
|
||
{"A": [0.60], "B": [0.40]},
|
||
execution_prices={"A": [100.0], "B": [80.0]},
|
||
)
|
||
|
||
positions = engine.bar_positions[0]
|
||
sizes = _sizes(positions)
|
||
# Proportions live in NOTIONAL space (A@100 vs B@80 → share ratio 1.2);
|
||
# fine lots (0.5% of each fill) keep the assertion meaningful.
|
||
assert sizes["A"] / sizes["B"] == pytest.approx(1.2, rel=2e-2)
|
||
notionals = (sizes["A"] * 100.0, sizes["B"] * 80.0)
|
||
assert notionals[0] / sum(notionals) == pytest.approx(0.60, rel=2e-2)
|
||
# Each fill commissioned under its OWN symbol's schedule.
|
||
assert positions["A"].entry_commission == pytest.approx(
|
||
sizes["A"] * 100.0 * 0.01
|
||
)
|
||
assert positions["B"].entry_commission == pytest.approx(
|
||
sizes["B"] * 80.0 * 0.02
|
||
)
|
||
assert engine.bar_capitals[0] >= 0.0
|
||
|
||
|
||
def test_rebalance_reduction_commits_and_increase_scales_to_fit():
|
||
"""#1274: reductions commit as planned; the open sleeve scales to fit."""
|
||
engine = _AdjustmentEngine(fee_rate=0.01)
|
||
_run_adjustments(engine, {"A": [0.25, 0.10], "B": [0.25, 0.90]})
|
||
|
||
first, second = engine.bar_positions
|
||
assert _sizes(first) == {"A": 2.5, "B": 2.5}
|
||
assert engine.bar_capitals[0] == 495.0
|
||
assert _sizes(second)["A"] == pytest.approx(0.995) # 10% of bar-2 equity 995
|
||
# B's increase is scaled by the one common factor to fit capital.
|
||
assert 2.5 < _sizes(second)["B"] < 9.0
|
||
assert engine.bar_capitals[1] >= 0.0
|
||
|
||
|
||
def test_rebalance_irrecoverably_infeasible_basket_fails_atomically():
|
||
"""#1274: when no scale fits — not even an empty open sleeve — abort."""
|
||
engine = _AdjustmentEngine()
|
||
with pytest.raises(ValueError, match="insufficient capital"):
|
||
_run_adjustments(
|
||
engine,
|
||
{"A": [-0.50, 0.0]},
|
||
execution_prices={"A": [100.0, 1000.0]},
|
||
)
|
||
# Bar 1's short is intact; the unrecoverable close released negative
|
||
# capital no scaling of opens (there are none) could repair. The abort
|
||
# committed nothing: exactly bar 1's fill exists, no bar-2 artifacts.
|
||
assert engine.capital == 500.0
|
||
position = engine.positions["A"]
|
||
assert (position.direction, position.size) == (-1, 5.0)
|
||
# Opens don't append trade records or adjustment events; the abort
|
||
# committed no close.
|
||
assert engine.trades == []
|
||
assert engine.adjustment_events == []
|
||
assert len(engine.bar_positions) == 1
|
||
|
||
@pytest.mark.parametrize(
|
||
("target_weight", "expected_position", "bar_capital", "trade_fees"),
|
||
[(0.0, None, 990.0, [10.0]), (-0.50, (-1, 4.975, 4.975), 487.525, [10.0, 9.95])])
|
||
def test_rebalance_full_zero_and_reversal_allocate_nonzero_fees(
|
||
target_weight, expected_position, bar_capital, trade_fees
|
||
):
|
||
engine = _AdjustmentEngine(fee_rate=0.01)
|
||
|
||
_run_adjustments(engine, {"A": [0.50, target_weight]})
|
||
|
||
position = engine.bar_positions[1].get("A")
|
||
actual_position = None if position is None else (position.direction, position.size, position.entry_commission)
|
||
assert actual_position == expected_position
|
||
assert engine.bar_capitals[1] == pytest.approx(bar_capital)
|
||
assert [trade.commission for trade in engine.trades] == pytest.approx(trade_fees)
|
||
assert engine.trades[0].exit_reason == "signal"
|
||
|
||
|
||
def test_rebalance_basket_is_independent_of_input_code_order():
|
||
weights = {"A": [0.50], "B": [0.50]}
|
||
first, second = _run_both_code_orders(_AdjustmentEngine, weights)
|
||
assert _sizes(first.bar_positions[0]) == {"A": 5.0, "B": 5.0}
|
||
assert first.bar_positions == second.bar_positions
|
||
assert [snapshot.capital for snapshot in first.equity_snapshots] == [
|
||
snapshot.capital for snapshot in second.equity_snapshots
|
||
]
|
||
|
||
|
||
def test_existing_rebalance_uses_each_symbol_rules_independent_of_code_order():
|
||
weights = {"A": [0.20, 0.35], "B": [0.20, 0.15]}
|
||
first, second = _run_both_code_orders(_SymbolRulesAdjustmentEngine, weights)
|
||
expected = [{"A": 2.0, "B": 2.0}, {"A": 3.0, "B": 1.25}]
|
||
assert [_sizes(state) for state in first.bar_positions] == expected
|
||
assert first.bar_positions == second.bar_positions
|
||
assert first.bar_capitals == pytest.approx([594.0, 566.5])
|
||
assert first.bar_capitals == second.bar_capitals
|
||
|
||
|
||
class _LifecycleEngine(ChinaAEngine):
|
||
def __init__(self, *, stop_before: bool = False):
|
||
super().__init__({"initial_cash": 1_000_000.0})
|
||
self.stop_before = stop_before
|
||
self.lifecycle: list[str] = []
|
||
|
||
def before_rebalance_bar(self, timestamp, data_map, codes):
|
||
self.lifecycle.append("pre")
|
||
return self.stop_before
|
||
|
||
def after_rebalance_bar(self, timestamp, data_map, codes):
|
||
self.lifecycle.append("post")
|
||
return False
|
||
|
||
def _execute_open_order(self, order, ts):
|
||
self.lifecycle.append("fill")
|
||
super()._execute_open_order(order, ts)
|
||
|
||
|
||
class _FractionalEngine(BaseEngine):
|
||
"""Frictionless engine used to expose target-vs-execution differences."""
|
||
|
||
def __init__(self, *, block_adds_after_first: bool = False):
|
||
super().__init__({"initial_cash": 1_000.0, "position_adjustment": "rebalance"})
|
||
self.block_adds_after_first = block_adds_after_first
|
||
|
||
def can_execute(self, symbol, direction, bar):
|
||
if (
|
||
self.block_adds_after_first
|
||
and direction != 0
|
||
and symbol in self.positions
|
||
):
|
||
return False
|
||
return True
|
||
|
||
def round_size(self, raw_size, price):
|
||
return raw_size
|
||
|
||
def calc_commission(self, size, price, direction, is_open):
|
||
return 0.0
|
||
|
||
def apply_slippage(self, price, direction):
|
||
return price
|
||
|
||
|
||
def _fractional_fixture(weights: list[float]):
|
||
dates = pd.bdate_range("2026-01-05", periods=len(weights))
|
||
bars = pd.DataFrame(
|
||
{"open": [100.0] * len(dates), "close": [100.0] * len(dates)},
|
||
index=dates,
|
||
)
|
||
close_df = pd.DataFrame({"AAPL.US": bars["close"]}, index=dates)
|
||
targets = pd.DataFrame({"AAPL.US": weights}, index=dates)
|
||
return dates, bars, close_df, targets
|
||
|
||
|
||
def test_same_direction_target_change_resizes_actual_position() -> None:
|
||
dates, bars, close_df, targets = _fractional_fixture([0.2, 0.8, 0.8])
|
||
engine = _FractionalEngine()
|
||
|
||
engine._execute_bars(
|
||
dates, {"AAPL.US": bars}, close_df, targets, ["AAPL.US"]
|
||
)
|
||
|
||
actual = engine._actual_positions_frame(["AAPL.US"])
|
||
assert actual.loc[dates[0], "AAPL.US"] == pytest.approx(0.2)
|
||
assert actual.loc[dates[1], "AAPL.US"] == pytest.approx(0.8)
|
||
# The terminal liquidation closes the full resized position, proving the
|
||
# engine held 8 shares rather than retaining the original 2 shares.
|
||
assert engine.trades[-1].size == pytest.approx(8.0)
|
||
|
||
|
||
def test_same_direction_target_reduction_partially_closes() -> None:
|
||
dates, bars, close_df, targets = _fractional_fixture([0.8, 0.2, 0.2])
|
||
engine = _FractionalEngine()
|
||
|
||
engine._execute_bars(
|
||
dates, {"AAPL.US": bars}, close_df, targets, ["AAPL.US"]
|
||
)
|
||
|
||
actual = engine._actual_positions_frame(["AAPL.US"])
|
||
assert actual.loc[dates[1], "AAPL.US"] == pytest.approx(0.2)
|
||
assert engine.trades[0].exit_reason == "target_rebalance"
|
||
assert engine.trades[0].size == pytest.approx(6.0)
|
||
assert engine.trades[-1].size == pytest.approx(2.0)
|
||
|
||
|
||
def test_positions_artifact_reports_fills_not_blocked_targets(tmp_path) -> None:
|
||
dates, bars, close_df, targets = _fractional_fixture([0.2, 0.8, 0.8])
|
||
engine = _FractionalEngine(block_adds_after_first=True)
|
||
engine._execute_bars(
|
||
dates, {"AAPL.US": bars}, close_df, targets, ["AAPL.US"]
|
||
)
|
||
equity = pd.Series(
|
||
[snapshot.equity for snapshot in engine.equity_snapshots], index=dates
|
||
)
|
||
benchmark_return = pd.Series(0.0, index=dates)
|
||
benchmark_equity = pd.Series(1_000.0, index=dates)
|
||
|
||
engine._write_artifacts(
|
||
tmp_path,
|
||
{"AAPL.US": bars},
|
||
dates,
|
||
equity,
|
||
benchmark_equity,
|
||
benchmark_return,
|
||
targets,
|
||
{},
|
||
["AAPL.US"],
|
||
)
|
||
|
||
actual_csv = pd.read_csv(tmp_path / "artifacts" / "positions.csv", index_col=0)
|
||
target_csv = pd.read_csv(
|
||
tmp_path / "artifacts" / "target_positions.csv", index_col=0
|
||
)
|
||
assert actual_csv.iloc[1]["AAPL.US"] == pytest.approx(0.2)
|
||
assert target_csv.iloc[1]["AAPL.US"] == pytest.approx(0.8)
|
||
|
||
|
||
def test_every_written_artifact_is_declared_to_the_runner(tmp_path) -> None:
|
||
"""An artifact the engine writes but the spec omits is invisible downstream.
|
||
|
||
``Runner`` builds its returned artifact map by walking
|
||
``_ARTIFACTS_SPEC``, so a file that is written and not declared exists on
|
||
disk while no caller can find it. Asserting the direction that matters
|
||
(written ⊆ declared) makes adding an artifact without registering it fail
|
||
here rather than silently.
|
||
"""
|
||
from src.core.runner import _ARTIFACTS_SPEC
|
||
|
||
dates, bars, close_df, targets = _fractional_fixture([0.2, 0.8, 0.8])
|
||
engine = _FractionalEngine(block_adds_after_first=True)
|
||
engine._execute_bars(dates, {"AAPL.US": bars}, close_df, targets, ["AAPL.US"])
|
||
equity = pd.Series(
|
||
[snapshot.equity for snapshot in engine.equity_snapshots], index=dates
|
||
)
|
||
engine._write_artifacts(
|
||
tmp_path,
|
||
{"AAPL.US": bars},
|
||
dates,
|
||
equity,
|
||
pd.Series(1_000.0, index=dates),
|
||
pd.Series(0.0, index=dates),
|
||
targets,
|
||
{},
|
||
["AAPL.US"],
|
||
)
|
||
|
||
declared = {
|
||
entry["path"].split("artifacts/", 1)[1]
|
||
for entry in _ARTIFACTS_SPEC["artifacts"].values()
|
||
if str(entry.get("path", "")).startswith("artifacts/")
|
||
}
|
||
# Per-symbol OHLCV copies are deliberately undeclared: their names depend on
|
||
# the universe, and they mirror the loader's input rather than a result.
|
||
written = {
|
||
path.name
|
||
for path in (tmp_path / "artifacts").glob("*.csv")
|
||
if not path.name.startswith("ohlcv_")
|
||
}
|
||
|
||
undeclared = sorted(written - declared)
|
||
assert not undeclared, f"written but not declared in _ARTIFACTS_SPEC: {undeclared}"
|
||
|
||
|
||
def _run_lifecycle(engine: _LifecycleEngine) -> None:
|
||
dates = pd.DatetimeIndex([pd.Timestamp("2026-01-02")])
|
||
frame = pd.DataFrame({"open": [100.0], "close": [100.0]}, index=dates)
|
||
engine._execute_bars(
|
||
dates,
|
||
{"TEST": frame},
|
||
frame[["close"]].rename(columns={"close": "TEST"}),
|
||
pd.DataFrame({"TEST": [1.0]}, index=dates),
|
||
["TEST"],
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(("stop_before", "expected"), [
|
||
(False, ["pre", "fill", "post"]), (True, ["pre"]),
|
||
])
|
||
def test_execute_bars_lifecycle_and_pre_fill_stop(
|
||
stop_before: bool, expected: list[str]
|
||
) -> None:
|
||
engine = _LifecycleEngine(stop_before=stop_before)
|
||
_run_lifecycle(engine)
|
||
assert engine.lifecycle == expected
|
||
assert len(engine.equity_snapshots) == 1
|
||
if stop_before:
|
||
assert engine.trades == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _align: signal alignment and normalization
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _simple_data_and_signals():
|
||
"""Build minimal data_map and signal_map for alignment tests."""
|
||
dates = pd.bdate_range("2025-01-01", periods=10)
|
||
df_a = pd.DataFrame(
|
||
{"close": np.linspace(10, 20, 10), "open": np.linspace(10, 20, 10)},
|
||
index=dates,
|
||
)
|
||
df_b = pd.DataFrame(
|
||
{"close": np.linspace(100, 110, 10), "open": np.linspace(100, 110, 10)},
|
||
index=dates,
|
||
)
|
||
data_map = {"A": df_a, "B": df_b}
|
||
|
||
sig_a = pd.Series(0.0, index=dates)
|
||
sig_a.iloc[3:] = 1.0
|
||
sig_b = pd.Series(0.0, index=dates)
|
||
sig_b.iloc[5:] = 1.0
|
||
signal_map = {"A": sig_a, "B": sig_b}
|
||
|
||
return data_map, signal_map, dates
|
||
|
||
|
||
class TestAlign:
|
||
def test_common_timezone_is_preserved(self) -> None:
|
||
dates = pd.date_range("2026-01-01", periods=3, freq="h", tz="UTC")
|
||
frame = pd.DataFrame({"open": [100.0] * 3, "close": [100.0] * 3}, index=dates)
|
||
signals = pd.Series([0.0, 1.0, 0.0], index=dates)
|
||
|
||
out_dates, close_df, _, pos_df, _ = _align(
|
||
{"BTC-USDT-PERP": frame}, {"BTC-USDT-PERP": signals}, ["BTC-USDT-PERP"]
|
||
)
|
||
|
||
assert str(out_dates.tz) == "UTC"
|
||
assert close_df.index.equals(dates)
|
||
assert pos_df.index.equals(dates)
|
||
|
||
def test_output_shapes(self) -> None:
|
||
data_map, signal_map, dates = _simple_data_and_signals()
|
||
out_dates, close_df, _, pos_df, ret_df = _align(data_map, signal_map, ["A", "B"])
|
||
assert len(out_dates) == len(dates)
|
||
assert close_df.shape == (len(dates), 2)
|
||
assert pos_df.shape == (len(dates), 2)
|
||
assert ret_df.shape == (len(dates), 2)
|
||
|
||
def test_signal_shifted_by_one(self) -> None:
|
||
"""Signal at bar i should produce position at bar i+1 (next-bar-open)."""
|
||
data_map, signal_map, dates = _simple_data_and_signals()
|
||
_, _, _, pos_df, _ = _align(data_map, signal_map, ["A", "B"])
|
||
# Signal A goes to 1.0 at index 3 → position should be 0 at index 3, non-zero at index 4
|
||
assert pos_df.at[dates[3], "A"] == 0.0
|
||
assert pos_df.at[dates[4], "A"] > 0.0
|
||
|
||
def test_positions_normalized(self) -> None:
|
||
"""Sum of abs(weights) should be <= 1.0 per row."""
|
||
data_map, signal_map, dates = _simple_data_and_signals()
|
||
_, _, _, pos_df, _ = _align(data_map, signal_map, ["A", "B"])
|
||
row_sums = pos_df.abs().sum(axis=1)
|
||
assert (row_sums <= 1.0 + 1e-10).all()
|
||
|
||
def test_signals_clipped(self) -> None:
|
||
"""Signals outside [-1, 1] should be clipped."""
|
||
dates = pd.bdate_range("2025-01-01", periods=5)
|
||
df = pd.DataFrame({"close": [100] * 5, "open": [100] * 5}, index=dates)
|
||
sig = pd.Series([0, 0, 2.0, -3.0, 0.5], index=dates)
|
||
data_map = {"X": df}
|
||
signal_map = {"X": sig}
|
||
_, _, _, pos_df, _ = _align(data_map, signal_map, ["X"])
|
||
# After shift, clipped values show up at indices 3 and 4
|
||
assert pos_df["X"].abs().max() <= 1.0 + 1e-10
|
||
|
||
def test_nan_signals_filled_zero(self) -> None:
|
||
dates = pd.bdate_range("2025-01-01", periods=5)
|
||
df = pd.DataFrame({"close": [100] * 5, "open": [100] * 5}, index=dates)
|
||
sig = pd.Series([np.nan, 1.0, np.nan, 0.5, np.nan], index=dates)
|
||
data_map = {"X": df}
|
||
signal_map = {"X": sig}
|
||
_, _, _, pos_df, _ = _align(data_map, signal_map, ["X"])
|
||
assert not pos_df.isna().any().any()
|
||
|
||
def test_close_ffill_bfill(self) -> None:
|
||
"""Missing close prices should be forward/backward filled."""
|
||
dates = pd.bdate_range("2025-01-01", periods=5)
|
||
df = pd.DataFrame(
|
||
{"close": [100, np.nan, np.nan, 110, 115], "open": [100] * 5},
|
||
index=dates,
|
||
)
|
||
sig = pd.Series([0, 1, 1, 1, 0], index=dates)
|
||
_, close_df, _, _, _ = _align({"X": df}, {"X": sig}, ["X"])
|
||
assert not close_df.isna().any().any()
|
||
|
||
def test_with_optimizer(self) -> None:
|
||
"""Optimizer callable gets applied."""
|
||
data_map, signal_map, dates = _simple_data_and_signals()
|
||
|
||
def dummy_optimizer(ret, pos, dates_arg):
|
||
return pos * 0.5 # halve everything
|
||
|
||
_, _, _, pos_df, _ = _align(data_map, signal_map, ["A", "B"], optimizer=dummy_optimizer)
|
||
# Positions should be smaller due to optimizer
|
||
_, _, _, pos_no_opt, _ = _align(data_map, signal_map, ["A", "B"])
|
||
assert pos_df.abs().sum().sum() <= pos_no_opt.abs().sum().sum() + 1e-10
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _load_optimizer
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestLoadOptimizer:
|
||
def test_no_optimizer(self) -> None:
|
||
assert _load_optimizer({}) is None
|
||
assert _load_optimizer({"optimizer": ""}) is None
|
||
|
||
def test_valid_optimizer(self) -> None:
|
||
opt = _load_optimizer({"optimizer": "risk_parity"})
|
||
assert opt is not None and callable(opt)
|
||
|
||
def test_invalid_optimizer_returns_none(self) -> None:
|
||
opt = _load_optimizer({"optimizer": "nonexistent_module_xyz"})
|
||
assert opt is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _close_position: PnL calculation
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestClosePosition:
|
||
def test_profitable_long(self) -> None:
|
||
engine = ChinaAEngine({"initial_cash": 1_000_000})
|
||
engine._bar_idx = 5
|
||
engine.positions["000001.SZ"] = Position(
|
||
"000001.SZ", 1, 15.0, pd.Timestamp("2025-01-02"), 1000.0, entry_bar_idx=0,
|
||
)
|
||
engine.capital = 985_000.0 # after buying
|
||
engine._close_position("000001.SZ", 16.0, pd.Timestamp("2025-01-10"), "signal")
|
||
|
||
assert "000001.SZ" not in engine.positions
|
||
assert len(engine.trades) == 1
|
||
t = engine.trades[0]
|
||
assert t.pnl == pytest.approx(1000.0) # 1000 × (16 - 15) = +1000
|
||
assert t.exit_reason == "signal"
|
||
assert t.holding_bars == 5
|
||
|
||
def test_losing_long(self) -> None:
|
||
engine = ChinaAEngine({"initial_cash": 1_000_000})
|
||
engine._bar_idx = 3
|
||
engine.positions["600519.SH"] = Position(
|
||
"600519.SH", 1, 1800.0, pd.Timestamp("2025-01-02"), 100.0, entry_bar_idx=0,
|
||
)
|
||
engine.capital = 820_000.0
|
||
engine._close_position("600519.SH", 1750.0, pd.Timestamp("2025-01-06"), "signal")
|
||
|
||
t = engine.trades[0]
|
||
assert t.pnl == pytest.approx(-5000.0) # 100 × (1750 - 1800) = -5000
|
||
assert t.direction == 1
|
||
|
||
def test_close_nonexistent_position_noop(self) -> None:
|
||
engine = ChinaAEngine({"initial_cash": 1_000_000})
|
||
engine._close_position("NOPE.SZ", 10.0, pd.Timestamp("2025-01-01"), "signal")
|
||
assert len(engine.trades) == 0
|
||
|
||
def test_capital_returned(self) -> None:
|
||
engine = ChinaAEngine({"initial_cash": 1_000_000})
|
||
engine._bar_idx = 1
|
||
engine.positions["000001.SZ"] = Position(
|
||
"000001.SZ", 1, 15.0, pd.Timestamp("2025-01-02"), 1000.0,
|
||
)
|
||
capital_before = 985_000.0
|
||
engine.capital = capital_before
|
||
engine._close_position("000001.SZ", 15.0, pd.Timestamp("2025-01-03"), "signal")
|
||
# Margin returned + 0 PnL - exit commission
|
||
assert engine.capital > capital_before # margin returned exceeds commission
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _calc_equity
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestCalcEquity:
|
||
def test_no_positions(self) -> None:
|
||
engine = ChinaAEngine({"initial_cash": 1_000_000})
|
||
dates = pd.DatetimeIndex([pd.Timestamp("2025-01-02")])
|
||
close_df = pd.DataFrame({"X": [15.0]}, index=dates)
|
||
eq = engine._calc_equity(close_df, dates[0])
|
||
assert eq == 1_000_000.0
|
||
|
||
def test_with_unrealized_gain(self) -> None:
|
||
engine = ChinaAEngine({"initial_cash": 1_000_000})
|
||
engine.capital = 985_000.0
|
||
engine.positions["X"] = Position("X", 1, 15.0, pd.Timestamp("2025-01-02"), 1000.0)
|
||
dates = pd.DatetimeIndex([pd.Timestamp("2025-01-03")])
|
||
close_df = pd.DataFrame({"X": [16.0]}, index=dates)
|
||
eq = engine._calc_equity(close_df, dates[0])
|
||
# capital + margin + unrealized = 985000 + (1000×15/1) + (1×1000×(16-15)) = 985000 + 15000 + 1000 = 1001000
|
||
assert eq == pytest.approx(1_001_000.0)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _safe_price
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSafePrice:
|
||
def test_returns_close_price(self) -> None:
|
||
dates = pd.DatetimeIndex([pd.Timestamp("2025-01-02")])
|
||
close_df = pd.DataFrame({"X": [15.5]}, index=dates)
|
||
assert BaseEngine._safe_price(close_df, dates[0], "X", 10.0) == 15.5
|
||
|
||
def test_fallback_on_missing_symbol(self) -> None:
|
||
dates = pd.DatetimeIndex([pd.Timestamp("2025-01-02")])
|
||
close_df = pd.DataFrame({"X": [15.5]}, index=dates)
|
||
assert BaseEngine._safe_price(close_df, dates[0], "MISSING", 10.0) == 10.0
|
||
|
||
def test_fallback_on_missing_timestamp(self) -> None:
|
||
dates = pd.DatetimeIndex([pd.Timestamp("2025-01-02")])
|
||
close_df = pd.DataFrame({"X": [15.5]}, index=dates)
|
||
assert BaseEngine._safe_price(close_df, pd.Timestamp("2025-06-01"), "X", 10.0) == 10.0
|
||
|
||
def test_fallback_on_nan(self) -> None:
|
||
dates = pd.DatetimeIndex([pd.Timestamp("2025-01-02")])
|
||
close_df = pd.DataFrame({"X": [np.nan]}, index=dates)
|
||
assert BaseEngine._safe_price(close_df, dates[0], "X", 10.0) == 10.0
|
||
|
||
|
||
def test_halted_position_marks_at_last_close_past_ffill_limit():
|
||
# #1318: a position held through a halt longer than the ffill limit used to
|
||
# be re-marked at entry price, producing a phantom drawdown mid-halt.
|
||
run_dates = pd.date_range("2026-01-02", periods=30, freq="B")
|
||
halt_dates = run_dates[:10]
|
||
halt_close = [100.0 + 5.0 * i for i in range(10)]
|
||
data_map = {
|
||
"HALT": pd.DataFrame({"open": halt_close, "close": halt_close}, index=halt_dates),
|
||
"RUN": pd.DataFrame({"open": 50.0, "close": 50.0}, index=run_dates),
|
||
}
|
||
signal_map = {
|
||
"HALT": pd.Series(0.5, index=halt_dates),
|
||
"RUN": pd.Series(0.0, index=run_dates),
|
||
}
|
||
dates, close_df, close_val_df, target_pos, _ = _align(data_map, signal_map, ["HALT", "RUN"])
|
||
|
||
engine = _AdjustmentEngine()
|
||
engine._execute_bars(
|
||
dates, data_map, close_df, target_pos, ["HALT", "RUN"],
|
||
close_val_df=close_val_df,
|
||
)
|
||
|
||
snaps = {s.timestamp: s.equity for s in engine.equity_snapshots}
|
||
marked_at_last_close = snaps[run_dates[9]]
|
||
# Deep into the halt the equity must not fall back to the entry-cost mark.
|
||
for bar in (14, 15, 20, 29):
|
||
assert snaps[run_dates[bar]] == pytest.approx(marked_at_last_close, rel=1e-9)
|
||
# The terminal forced liquidation also marks at the last traded close, so
|
||
# the halt's unrealized PnL survives into the final equity.
|
||
assert engine.equity_snapshots[-1].equity == pytest.approx(marked_at_last_close, rel=1e-6)
|
||
|
||
|
||
def test_rebalance_scaled_away_sleeve_is_recorded_as_plan_rejection():
|
||
"""#1274: a sleeve rounded to zero by scaling leaves an audit record.
|
||
|
||
A fills 9 of its 10 target shares; B's one-lot fill scales below one lot
|
||
and vanishes — run-card diagnostics must see it as an insufficient_capital
|
||
rejection (#1470: it was a real order at full scale), not silence and not
|
||
a lot-rounding failure.
|
||
"""
|
||
engine = _SymbolRulesAdjustmentEngine()
|
||
_run_adjustments(engine, {"A": [1.0], "B": [0.025]}) # B: exactly one 0.25 lot
|
||
|
||
sizes = _sizes(engine.bar_positions[0])
|
||
assert sizes == {"A": 9.0} # scaled to fit; B's leg dropped entirely
|
||
assert engine.plan_rejections[("B", "insufficient_capital")] == 1
|
||
assert ("B", "zero_size") not in engine.plan_rejections
|
||
|
||
|
||
def test_rebalance_sleeve_below_one_lot_at_full_scale_is_the_lot_rule():
|
||
"""A target that never reached one lot is zero_size, not a cash finding."""
|
||
engine = _SymbolRulesAdjustmentEngine()
|
||
_run_adjustments(engine, {"A": [1.0], "B": [0.003]}) # B: 0.03 shares, lot 0.25
|
||
|
||
assert _sizes(engine.bar_positions[0]) == {"A": 9.0}
|
||
assert engine.plan_rejections[("B", "zero_size")] == 1
|
||
assert ("B", "insufficient_capital") not in engine.plan_rejections
|
||
assert engine.bar_capitals[0] >= 0.0
|
||
|
||
|
||
class _FundingDebitAdjustmentEngine(_AdjustmentEngine):
|
||
"""Debits a fee after every bar, as CompositeEngine's crypto funding does."""
|
||
|
||
def __init__(self, **overrides):
|
||
super().__init__(**overrides)
|
||
self.rejected: list[tuple[str, str]] = []
|
||
|
||
def after_rebalance_bar(self, timestamp, data_map, codes):
|
||
self.capital -= float(self.config.get("debit", 0.0))
|
||
return super().after_rebalance_bar(timestamp, data_map, codes)
|
||
|
||
def _on_plan_rejected(self, symbol, reason, timestamp):
|
||
self.rejected.append((symbol, reason))
|
||
|
||
|
||
def test_rebalance_with_negative_cash_drops_the_opens_and_keeps_running():
|
||
"""#1542's rebalance sibling: when even an empty open sleeve does not fit
|
||
(cash below zero, nothing reduced to release it), the run used to abort with
|
||
'insufficient capital for position rebalance'. The opens are dropped and
|
||
reported; the rest of the bar and the run go on.
|
||
|
||
A takes all 1,000 of cash on bar 0 and the debit leaves -30. A's weight on
|
||
the later bars keeps it at exactly 10 shares, so nothing is released.
|
||
"""
|
||
engine = _FundingDebitAdjustmentEngine(debit=30.0)
|
||
keep_a = 1_000.0 / 970.0
|
||
_run_adjustments(engine, {"A": [1.0, keep_a, keep_a], "B": [0.0, 0.2, 0.2]})
|
||
|
||
assert engine.bar_capitals[0] == pytest.approx(-30.0)
|
||
assert _sizes(engine.bar_positions[1]) == {"A": 10.0}
|
||
assert ("B", "insufficient_capital") in engine.rejected
|
||
assert len(engine.bar_positions) == 3 # the run reached its last bar
|
||
|
||
|
||
def test_rebalance_with_negative_cash_still_executes_its_reductions():
|
||
"""Dropping the opens must not drop the reductions: they only release cash.
|
||
|
||
Cash is -300 after bar 0 (equity 700); trimming A to 9 shares releases 100,
|
||
not enough to bring cash to zero, so no open sleeve fits. The trim happens.
|
||
"""
|
||
engine = _FundingDebitAdjustmentEngine(debit=300.0)
|
||
_run_adjustments(engine, {"A": [1.0, 900.0 / 700.0], "B": [0.0, 0.1]})
|
||
|
||
assert _sizes(engine.bar_positions[1]) == {"A": 9.0}
|
||
assert ("B", "insufficient_capital") in engine.rejected
|
||
|
||
|
||
@pytest.mark.parametrize("units", [("us", "us"), ("ns", "us"), ("s", "ms")])
|
||
def test_align_is_the_same_at_any_index_resolution(units: tuple[str, str]) -> None:
|
||
"""_align works on int64 epochs; a duckdb local source arrives as datetime64[us].
|
||
|
||
Read as nanoseconds, a microsecond index put the whole run in 1970, and beside
|
||
a nanosecond source it matched none of that symbol's bars (an all-NaN column).
|
||
"""
|
||
days = pd.bdate_range("2026-01-02", periods=5)
|
||
closes = np.array([10.0, 11.0, 12.0, 13.0, 14.0])
|
||
|
||
def run(unit_a: str, unit_b: str):
|
||
idx_a, idx_b = days.as_unit(unit_a), days.as_unit(unit_b)
|
||
data_map = {
|
||
"A": pd.DataFrame({"open": closes, "high": closes, "low": closes, "close": closes}, index=idx_a),
|
||
"B": pd.DataFrame({"open": closes, "high": closes, "low": closes, "close": closes * 2}, index=idx_b),
|
||
}
|
||
signal_map = {"A": pd.Series([1.0, 0, 1, 0, 1], index=idx_a), "B": pd.Series(0.5, index=idx_b)}
|
||
return _align(data_map, signal_map, ["A", "B"])
|
||
|
||
expected_dates, expected_close, _, expected_pos, _ = run("ns", "ns")
|
||
dates, close_df, _, pos_df, _ = run(*units)
|
||
|
||
assert list(dates) == list(expected_dates)
|
||
np.testing.assert_array_equal(close_df.to_numpy(), expected_close.to_numpy())
|
||
np.testing.assert_array_equal(pos_df.to_numpy(), expected_pos.to_numpy())
|