royy/chinese-equity-quant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Raise coverage threshold to 95% and expand test coverage

Browse Source 
- pyproject.toml: fail_under 80 → 95
- test_alpha: +79 lines
- test_cli_workflow: +226 lines
- test_derived: +121 lines
- test_downloader_contracts: +169 lines
- test_features: +16 lines
- test_minute_downloader: +81 lines
- test_portfolio: +208 lines
This commit is contained in:
Yuxuan Yan
2026-06-16 21:10:30 +08:00
parent b5c8c0b8da
commit 528620b271
8 changed files with 898 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_portfolio.py
+208
View File
@@ -13,12 +13,17 @@ from pipeline.portfolio.market_rules import (
Board,
LimitStatus,
MarketRule,
_to_date,
compute_limit_status,
detect_board,
)
from pipeline.portfolio.research import evaluate_portfolio
from pipeline.portfolio.constraints import (
TradeConstraint,
available_constraints,
get_constraint,
PriceLimitConstraint,
register_constraint,
SuspensionConstraint,
VolumeCapConstraint,
)
@@ -82,6 +87,7 @@ def test_detect_board():
assert detect_board("sh688981") == Board.STAR
assert detect_board("sz300750") == Board.CHINEXT
assert detect_board("bj830000") == Board.UNKNOWN
assert detect_board("sh") == Board.UNKNOWN
# --- MarketRule date transitions ---------------------------------------------
@@ -120,6 +126,26 @@ def test_get_rules_vectorized():
assert list(limit) == [0.10, 0.20, 0.20]
def test_market_rule_date_coercion_unknown_board_and_st_vector_override():
rules = MarketRule()
assert _to_date(dt.datetime(2024, 1, 2, 15, 0)) == dt.date(2024, 1, 2)
assert _to_date(pd.Timestamp("2024-01-03 09:30")) == dt.date(2024, 1, 3)
assert _to_date("2024-01-04 10:00:00") == dt.date(2024, 1, 4)
unknown_rule = rules.get_rule("xx999999", "2024-01-02")
assert unknown_rule.minimum_open_size == 100
assert unknown_rule.share_increment == 100
assert unknown_rule.price_limit_pct == 0.10
_, _, _, limit = rules.get_rules_vectorized(
np.array(["sh600000", "sh600000"], dtype=object),
"2024-01-02",
np.array([0, 1]),
)
assert limit.tolist() == [0.10, 0.05]
def test_compute_limit_status():
price = np.array([110.0, 90.0, 100.0])
preclose = np.array([100.0, 100.0, 100.0])
@@ -130,6 +156,20 @@ def test_compute_limit_status():
assert status[2] == LimitStatus.NORMAL.value
def test_compute_limit_status_treats_bad_preclose_as_normal():
status = compute_limit_status(
price=np.array([np.nan, 110.0, 90.0]),
preclose=np.array([100.0, np.nan, 0.0]),
limit_pct=np.array([0.10, 0.10, 0.10]),
)
assert status.tolist() == [
LimitStatus.NORMAL.value,
LimitStatus.NORMAL.value,
LimitStatus.NORMAL.value,
]
# --- continuous targets ------------------------------------------------------
def test_continuous_targets_normalization():
@@ -295,6 +335,70 @@ def test_repair_scales_to_4000_names():
assert abs(gross - B) <= 0.03 * B
def test_repair_handles_empty_input():
result = repair_exposure(
np.array([], dtype=np.int64),
np.array([], dtype=float),
np.array([], dtype=float),
np.array([], dtype=np.int64),
np.array([], dtype=np.int64),
np.array([], dtype=np.int64),
)
assert result.dtype == np.int64
assert result.tolist() == []
def test_repair_respects_max_iters_and_zero_increment_noop():
q_round = np.array([100, -100], dtype=np.int64)
q_target = np.array([300.0, -300.0])
price = np.array([10.0, 10.0])
min_open = np.array([100, 100])
prev = np.zeros(2, dtype=np.int64)
capped = repair_exposure(
q_round,
q_target,
price,
increment=np.array([1, 1]),
min_open=min_open,
prev_shares=prev,
booksize=10_000.0,
gross_tol=0.0,
max_iters=0,
)
zero_increment = repair_exposure(
q_round,
q_target,
price,
increment=np.array([0, 0]),
min_open=min_open,
prev_shares=prev,
booksize=0.0,
net_tol=0.0,
gross_tol=0.0,
)
assert capped.tolist() == [100, -100]
assert zero_increment.tolist() == [100, -100]
def test_repair_gross_growth_obeys_net_band():
pos = repair_exposure(
q_round=np.array([100], dtype=np.int64),
q_target=np.array([300.0]),
price=np.array([10.0]),
increment=np.array([1]),
min_open=np.array([100]),
prev_shares=np.array([0], dtype=np.int64),
booksize=2_000.0,
net_tol=0.5,
gross_tol=0.0,
)
assert pos.tolist() == [100]
# --- construct_positions -----------------------------------------------------
def test_construct_positions_schema():
@@ -497,8 +601,81 @@ def test_constraints_compose_repeatably_regardless_of_order():
assert np.array_equal(first_order.cost, reversed_order.cost)
def test_constraint_registry_and_default_adjust_targets():
class _NoopConstraint(TradeConstraint):
name = "_coverage_noop_constraint"
def delta_bounds(self, ctx):
return np.zeros(1), np.ones(1)
registered = register_constraint(_NoopConstraint)
assert registered is _NoopConstraint
assert "_coverage_noop_constraint" in available_constraints()
assert isinstance(get_constraint("_coverage_noop_constraint"), _NoopConstraint)
assert get_constraint("_coverage_noop_constraint").adjust_targets(object()) is None
with np.testing.assert_raises(KeyError):
get_constraint("_missing_constraint")
with np.testing.assert_raises(TypeError):
register_constraint(object) # type: ignore[arg-type]
with np.testing.assert_raises(ValueError):
class _NoNameConstraint(TradeConstraint):
def delta_bounds(self, ctx):
return np.zeros(1), np.ones(1)
register_constraint(_NoNameConstraint)
with np.testing.assert_raises(ValueError):
class _DuplicateConstraint(TradeConstraint):
name = "_coverage_noop_constraint"
def delta_bounds(self, ctx):
return np.zeros(1), np.ones(1)
register_constraint(_DuplicateConstraint)
# --- ReferenceSimulator ------------------------------------------------------
def test_simulator_applies_constraint_target_adjustment():
class _HalveTarget(TradeConstraint):
name = "_halve_target"
def adjust_targets(self, ctx):
return ctx.target_shares // 2
def delta_bounds(self, ctx):
return np.full(len(ctx.target_shares), -np.inf), np.full(len(ctx.target_shares), np.inf)
sl = _slice(1, price=np.array([10.0]))
ctx = TradeContext(np.array([0], np.int64), np.array([100], np.int64), sl, 1e6)
result = ReferenceSimulator(constraints=[_HalveTarget()]).fill(ctx)
assert result.traded_shares.tolist() == [50]
assert result.realized_shares.tolist() == [50]
def test_simulator_empty_positions_uses_default_booksize():
data = pd.DataFrame({
"symbol_id": ["sh600000"],
"date": [pd.Timestamp("2024-01-02")],
"open": [10.0],
"close": [10.0],
"preclose": [10.0],
"amount": [1e9],
"tradestatus": [1],
"isST": [0],
})
positions = pd.DataFrame(columns=POSITION_COLUMNS)
fills, pnl = ReferenceSimulator().run(positions, data)
assert list(fills.columns) == FILL_COLUMNS
assert list(pnl.columns) == PNL_COLUMNS
assert fills.empty
assert pnl.empty
def test_simulator_next_open_and_blocked_buy_holds_prev():
data = _make_data(n_days=15)
weights = _make_weights(data)
@@ -749,3 +926,34 @@ def test_evaluate_portfolio_excludes_signal_without_forward_return():
metrics = evaluate_portfolio(positions, data)
assert metrics["n_dates"] == 1
def test_evaluate_portfolio_empty_and_single_return_paths():
empty_metrics = evaluate_portfolio(
pd.DataFrame(columns=POSITION_COLUMNS),
pd.DataFrame(columns=["symbol_id", "date", "open"]),
)
assert empty_metrics["n_dates"] == 0
assert empty_metrics["cumulative_return"] == 0.0
dates = pd.date_range("2024-01-01", periods=3)
data = pd.DataFrame([
{"symbol_id": "sh600000", "date": d, "open": price}
for d, price in zip(dates, [100.0, 100.0, 110.0])
])
positions = pd.DataFrame({
"symbol_id": ["sh600000"],
"date": [dates[0]],
"portfolio_name": ["single"],
"target_weight": [1.0],
"target_value": [1000.0],
"target_shares": [10.0],
"position_shares": [10],
"position_value": [1000.0],
"price": [100.0],
})
single_metrics = evaluate_portfolio(positions, data)
assert single_metrics["n_dates"] == 1
assert single_metrics["cumulative_return"] == 0.0