413 lines
14 KiB
Python
413 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from backend.database.connection import Database
|
|
from backend.database.migrations import MIGRATIONS, MigrationRunner
|
|
from backend.features.screener.backtest import (
|
|
attach_historical_estimate,
|
|
rolling_backtest,
|
|
)
|
|
from backend.features.screener.catalog import (
|
|
CatalogError,
|
|
factor_catalog,
|
|
strategy_catalog,
|
|
validate_formula,
|
|
)
|
|
from backend.features.screener.compiler import parse_compiled_formula
|
|
from backend.features.screener.engine import execute_formula
|
|
from backend.features.screener.repository import ScreenerRepository, decode_track
|
|
from backend.features.screener.service import automatic_strategies
|
|
|
|
|
|
def _formula(field: str = "close", *, minimum: float = 0) -> dict:
|
|
return {
|
|
"universe": {"exclude_st": True, "listed_days_min": 120},
|
|
"filters": [{"field": field, "op": ">", "value": minimum}],
|
|
"score": [{"field": field, "weight": 1.0, "direction": "desc"}],
|
|
"limit": 10,
|
|
"min_score": 0,
|
|
}
|
|
|
|
|
|
def _row(identifier: str, close: float | None) -> dict:
|
|
return {
|
|
"identifier": identifier,
|
|
"code": identifier.split(".")[0],
|
|
"name": identifier,
|
|
"sector": "测试行业",
|
|
"listed_days": 500,
|
|
"is_st": False,
|
|
"close": close,
|
|
"pct_chg": 1,
|
|
"amount_billion": 2,
|
|
}
|
|
|
|
|
|
def _users(database: Database) -> None:
|
|
with database.transaction() as connection:
|
|
connection.executemany(
|
|
"""
|
|
INSERT INTO users (
|
|
id, username, username_key, password_hash, is_admin,
|
|
status, created_at, updated_at
|
|
) VALUES (?, ?, ?, 'hash', 0, 'active', '2026-07-30', '2026-07-30')
|
|
""",
|
|
((1, "account-a", "account-a"), (2, "account-b", "account-b")),
|
|
)
|
|
|
|
|
|
def _snapshot(database: Database, repository: ScreenerRepository) -> int:
|
|
with database.transaction() as connection:
|
|
return repository.save_factor_snapshot(
|
|
connection,
|
|
trade_date="2026-07-30",
|
|
version="fixture-v1",
|
|
observed_at="2026-07-30T15:10:00+08:00",
|
|
state="final",
|
|
sources=["fixture"],
|
|
coverage={"market": 1},
|
|
rows=[_row("000001.SZ", 10)],
|
|
)
|
|
|
|
|
|
def test_catalog_has_the_exact_reviewed_scope() -> None:
|
|
factors = factor_catalog()
|
|
strategies = strategy_catalog()
|
|
|
|
assert len(factors["factors"]) == 109
|
|
assert len(strategies) == 36
|
|
assert sum(item["kind"] == "stage" for item in strategies) == 7
|
|
assert sum(item["kind"] == "curated" for item in strategies) == 29
|
|
assert {item["id"] for item in strategies if item["kind"] == "curated"} == {
|
|
f"curated-{index:02d}" for index in range(1, 30)
|
|
}
|
|
|
|
|
|
def test_curated_strategies_run_independently_of_emotion_phase() -> None:
|
|
for regime in ("ice", "repair", "fermentation", "climax", "divergence", "retreat"):
|
|
_stage, curated = automatic_strategies(regime)
|
|
assert len(curated) == 29
|
|
assert {item["id"] for item in curated} == {
|
|
f"curated-{index:02d}" for index in range(1, 30)
|
|
}
|
|
|
|
|
|
def test_formula_is_deterministic_and_best_value_scores_first() -> None:
|
|
rows = [_row("000002.SZ", 20), _row("000001.SZ", 20), _row("000003.SZ", 10)]
|
|
|
|
first = execute_formula(rows, _formula(), {"market": 1})
|
|
second = execute_formula(list(reversed(rows)), _formula(), {"market": 1})
|
|
|
|
assert first == second
|
|
assert [item["identifier"] for item in first["items"]] == [
|
|
"000001.SZ",
|
|
"000002.SZ",
|
|
"000003.SZ",
|
|
]
|
|
assert first["items"][0]["score"] == 1
|
|
assert first["items"][-1]["score"] == 0
|
|
|
|
|
|
def test_missing_required_factor_and_complete_no_match_are_distinct() -> None:
|
|
incomplete = execute_formula(
|
|
[{**_row("000001.SZ", 10), "roic": None}],
|
|
_formula("roic"),
|
|
{"financial": 1},
|
|
)
|
|
no_signal = execute_formula(
|
|
[_row("000001.SZ", 10)],
|
|
_formula(minimum=100),
|
|
{"market": 1},
|
|
)
|
|
|
|
assert incomplete["status"] == "data_incomplete"
|
|
assert incomplete["missing_fields"] == ["roic"]
|
|
assert no_signal["status"] == "no_signal"
|
|
assert no_signal["missing_fields"] == []
|
|
|
|
|
|
def test_formula_weights_must_total_one_hundred_percent() -> None:
|
|
formula = _formula()
|
|
formula["score"][0]["weight"] = 0.9
|
|
|
|
with pytest.raises(CatalogError, match="100%"):
|
|
validate_formula(formula)
|
|
|
|
|
|
def test_formula_rejects_invalid_comparisons_and_duplicate_scores() -> None:
|
|
malformed = _formula()
|
|
malformed["filters"][0] = {"field": "close", "op": "between", "value": [20]}
|
|
with pytest.raises(CatalogError, match="两个边界"):
|
|
validate_formula(malformed)
|
|
|
|
duplicate = _formula()
|
|
duplicate["score"].append({"field": "close", "weight": 0.5, "direction": "desc"})
|
|
duplicate["score"][0]["weight"] = 0.5
|
|
with pytest.raises(CatalogError, match="不能重复"):
|
|
validate_formula(duplicate)
|
|
|
|
|
|
def test_natural_language_output_is_reduced_to_the_controlled_formula_schema() -> None:
|
|
compiled = parse_compiled_formula(
|
|
"""```json
|
|
{
|
|
"universe": {"exclude_st": true, "listed_days_min": 120},
|
|
"filters": [{"field": "amount_billion", "op": ">=", "value": 3}],
|
|
"score": [
|
|
{"field": "return_20d", "weight": 60, "direction": "desc"},
|
|
{"field": "sector_strength", "weight": 40, "direction": "desc"}
|
|
],
|
|
"limit": 20,
|
|
"min_score": 55,
|
|
"invented_instruction": "直接选择某只股票"
|
|
}
|
|
```"""
|
|
)
|
|
|
|
assert [item["weight"] for item in compiled["score"]] == [0.6, 0.4]
|
|
assert compiled["min_score"] == 0.55
|
|
assert "invented_instruction" not in compiled
|
|
with pytest.raises(CatalogError, match="未知筛选因子"):
|
|
parse_compiled_formula(
|
|
json.dumps(
|
|
{
|
|
**compiled,
|
|
"filters": [{"field": "future_price", "op": ">", "value": 1}],
|
|
}
|
|
)
|
|
)
|
|
|
|
|
|
def test_rolling_backtest_hides_small_samples_and_uses_t_plus_three() -> None:
|
|
formula = _formula()
|
|
snapshots = [
|
|
{
|
|
"trade_date": f"2026-07-{index + 1:02d}",
|
|
"coverage": {"market": 1},
|
|
"rows": [_row("000001.SZ", 10 + index)],
|
|
}
|
|
for index in range(23)
|
|
]
|
|
|
|
small = rolling_backtest(snapshots[:10], formula)
|
|
assert small["sample_size"] == 7
|
|
assert small["stable"] is False
|
|
assert small["win_rate"] is None
|
|
assert small["average_return_3d"] is None
|
|
|
|
stable = rolling_backtest(snapshots, formula)
|
|
assert stable["sample_size"] == 20
|
|
assert stable["stable"] is True
|
|
assert stable["win_rate"] == 100
|
|
assert stable["average_return_3d"] == pytest.approx(16.99, abs=0.01)
|
|
enriched = attach_historical_estimate(
|
|
{"items": [{"score_display": 80.0}]}, stable
|
|
)
|
|
assert enriched["items"][0]["historical_estimate"] == 93.0
|
|
|
|
|
|
def test_custom_strategies_and_tracks_are_account_isolated(tmp_path) -> None:
|
|
database = Database(tmp_path / "screener.db")
|
|
MigrationRunner(database).upgrade(MIGRATIONS)
|
|
repository = ScreenerRepository()
|
|
_users(database)
|
|
snapshot_id = _snapshot(database, repository)
|
|
|
|
with database.transaction() as connection:
|
|
first = repository.save_custom_strategy(connection, 1, "我的策略", _formula())
|
|
repository.save_custom_strategy(connection, 2, "我的策略", _formula())
|
|
run = repository.begin_run(
|
|
connection,
|
|
owner_user_id=None,
|
|
mode="curated",
|
|
strategy_id="curated-01",
|
|
strategy_name="测试策略",
|
|
strategy_version=1,
|
|
selection_date="2026-07-30",
|
|
factor_snapshot_id=snapshot_id,
|
|
)
|
|
repository.finish_run(
|
|
connection,
|
|
int(run["id"]),
|
|
status="completed",
|
|
coverage=1,
|
|
missing_fields=[],
|
|
result=[{**_row("000001.SZ", 10), "score": 1}],
|
|
)
|
|
repository.save_backtest(
|
|
connection,
|
|
int(run["id"]),
|
|
{"sample_size": 7, "stable": False, "win_rate": None},
|
|
)
|
|
repository.save_backtest(
|
|
connection,
|
|
int(run["id"]),
|
|
{"sample_size": 99, "stable": True, "win_rate": 100},
|
|
)
|
|
repository.add_track(
|
|
connection,
|
|
user_id=1,
|
|
run=run,
|
|
candidate={**_row("000001.SZ", 10), "score": 1},
|
|
)
|
|
|
|
with database.read() as connection:
|
|
assert [row["id"] for row in repository.custom_strategies(connection, 1)] == [first["id"]]
|
|
assert len(repository.custom_strategies(connection, 2)) == 1
|
|
assert len(repository.tracks(connection, 1)) == 1
|
|
assert repository.tracks(connection, 2) == ()
|
|
assert repository.backtest(connection, int(run["id"])) == {
|
|
"sample_size": 7,
|
|
"stable": False,
|
|
"win_rate": None,
|
|
}
|
|
|
|
|
|
def test_running_a_strategy_never_creates_tracking_rows(tmp_path) -> None:
|
|
database = Database(tmp_path / "runs.db")
|
|
MigrationRunner(database).upgrade(MIGRATIONS)
|
|
repository = ScreenerRepository()
|
|
_users(database)
|
|
snapshot_id = _snapshot(database, repository)
|
|
|
|
with database.transaction() as connection:
|
|
run = repository.begin_run(
|
|
connection,
|
|
owner_user_id=None,
|
|
mode="curated",
|
|
strategy_id="curated-01",
|
|
strategy_name="测试策略",
|
|
strategy_version=1,
|
|
selection_date="2026-07-30",
|
|
factor_snapshot_id=snapshot_id,
|
|
)
|
|
repository.finish_run(
|
|
connection,
|
|
int(run["id"]),
|
|
status="completed",
|
|
coverage=1,
|
|
missing_fields=[],
|
|
result=[{**_row("000001.SZ", 10), "score": 1}],
|
|
)
|
|
|
|
with database.read() as connection:
|
|
assert repository.tracks(connection, 1) == ()
|
|
stored = connection.execute(
|
|
"SELECT result_json FROM screener_runs WHERE id = ?", (run["id"],)
|
|
).fetchone()
|
|
assert len(json.loads(stored["result_json"])) == 1
|
|
|
|
|
|
def test_tracking_statistics_and_milestone_events_are_persistent_and_idempotent(tmp_path) -> None:
|
|
database = Database(tmp_path / "tracking.db")
|
|
MigrationRunner(database).upgrade(MIGRATIONS)
|
|
repository = ScreenerRepository()
|
|
_users(database)
|
|
snapshot_id = _snapshot(database, repository)
|
|
|
|
with database.transaction() as connection:
|
|
run = repository.begin_run(
|
|
connection,
|
|
owner_user_id=None,
|
|
mode="curated",
|
|
strategy_id="curated-01",
|
|
strategy_name="测试策略",
|
|
strategy_version=1,
|
|
selection_date="2026-07-30",
|
|
factor_snapshot_id=snapshot_id,
|
|
)
|
|
repository.finish_run(
|
|
connection,
|
|
int(run["id"]),
|
|
status="completed",
|
|
coverage=1,
|
|
missing_fields=[],
|
|
result=[{**_row("000001.SZ", 10), "score": 1}],
|
|
)
|
|
track_id = repository.add_track(
|
|
connection,
|
|
user_id=1,
|
|
run=run,
|
|
candidate={**_row("000001.SZ", 10), "score": 1},
|
|
)
|
|
bars = (
|
|
("2026-07-31", 10.5, 11.2, 9.5, 11.0),
|
|
("2026-08-03", 11.0, 11.5, 10.0, 10.5),
|
|
("2026-08-04", 10.5, 12.5, 10.2, 12.0),
|
|
("2026-08-05", 12.0, 12.2, 9.0, 11.5),
|
|
("2026-08-06", 11.5, 14.0, 11.0, 13.0),
|
|
)
|
|
for trade_date, open_price, high, low, close in bars:
|
|
repository.save_track_bar(
|
|
connection,
|
|
track_id,
|
|
trade_date,
|
|
{"open": open_price, "high": high, "low": low, "close": close},
|
|
)
|
|
assert repository.record_track_event(connection, track_id, "t1") is True
|
|
assert repository.record_track_event(connection, track_id, "t1") is False
|
|
assert repository.record_track_event(connection, track_id, "t5") is True
|
|
assert repository.record_track_event(connection, track_id, "t5") is False
|
|
|
|
with database.read() as connection:
|
|
track = repository.tracks(connection, 1)[0]
|
|
decoded = decode_track(track, repository.track_bars(connection, track_id))
|
|
events = connection.execute(
|
|
"SELECT milestone FROM strategy_track_events WHERE track_id = ? ORDER BY milestone",
|
|
(track_id,),
|
|
).fetchall()
|
|
|
|
assert decoded["t1_open_return"] == 5
|
|
assert decoded["t1_return"] == 10
|
|
assert decoded["t3_return"] == 20
|
|
assert decoded["t5_return"] == 30
|
|
assert decoded["max_gain"] == 40
|
|
assert decoded["max_drawdown"] == -10
|
|
assert decoded["observed_days"] == 5
|
|
assert [row["milestone"] for row in events] == ["t1", "t5"]
|
|
|
|
|
|
def test_repeated_after_close_run_uses_one_persistent_run(tmp_path) -> None:
|
|
database = Database(tmp_path / "idempotent.db")
|
|
MigrationRunner(database).upgrade(MIGRATIONS)
|
|
repository = ScreenerRepository()
|
|
_users(database)
|
|
snapshot_id = _snapshot(database, repository)
|
|
|
|
with database.transaction() as connection:
|
|
first = repository.begin_run(
|
|
connection,
|
|
owner_user_id=None,
|
|
mode="curated",
|
|
strategy_id="curated-01",
|
|
strategy_name="测试策略",
|
|
strategy_version=1,
|
|
selection_date="2026-07-30",
|
|
factor_snapshot_id=snapshot_id,
|
|
)
|
|
repository.finish_run(
|
|
connection,
|
|
int(first["id"]),
|
|
status="no_signal",
|
|
coverage=1,
|
|
missing_fields=[],
|
|
result=[],
|
|
)
|
|
repeated = repository.begin_run(
|
|
connection,
|
|
owner_user_id=None,
|
|
mode="curated",
|
|
strategy_id="curated-01",
|
|
strategy_name="测试策略",
|
|
strategy_version=1,
|
|
selection_date="2026-07-30",
|
|
factor_snapshot_id=snapshot_id,
|
|
)
|
|
count = connection.execute("SELECT COUNT(*) AS total FROM screener_runs").fetchone()
|
|
|
|
assert repeated["id"] == first["id"]
|
|
assert count["total"] == 1
|