360 lines
15 KiB
Python
360 lines
15 KiB
Python
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from database import ReviewDatabase
|
|
from screener import (
|
|
ADVANCED_CURATED_STRATEGIES,
|
|
CURATED_STRATEGIES,
|
|
FACTOR_FIELDS,
|
|
FACTOR_GROUPS,
|
|
ScreenerEngine,
|
|
_broken_reversal_metrics,
|
|
_risk_flags,
|
|
_rsi,
|
|
_quarter_periods,
|
|
)
|
|
from server import automatic_screener_jobs
|
|
|
|
|
|
class CuratedScreenerTests(unittest.TestCase):
|
|
def test_curated_library_contains_original_and_advanced_strategies(self):
|
|
self.assertEqual(13, len(ADVANCED_CURATED_STRATEGIES))
|
|
self.assertEqual(23, len(CURATED_STRATEGIES))
|
|
self.assertEqual(23, len({item["name"] for item in CURATED_STRATEGIES}))
|
|
self.assertTrue(
|
|
{"行业动量轮动", "主力资金行业流入"}.issubset(
|
|
{item["name"] for item in CURATED_STRATEGIES}
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(item["formula"]["meta"]["library"] == "curated" for item in CURATED_STRATEGIES)
|
|
)
|
|
|
|
def test_every_curated_strategy_explains_environment_and_failure_risk(self):
|
|
for strategy in CURATED_STRATEGIES:
|
|
meta = strategy["formula"]["meta"]
|
|
self.assertTrue(meta.get("suitable_environment"), strategy["name"])
|
|
self.assertTrue(meta.get("failure_risk"), strategy["name"])
|
|
self.assertNotIn("emotion_gate", meta, strategy["name"])
|
|
|
|
def test_automatic_curated_jobs_are_not_filtered_by_market_regime(self):
|
|
strategies = [
|
|
{
|
|
"name": "阶段策略",
|
|
"regimes": ["retreat"],
|
|
"formula": {"meta": {"library": "stage"}},
|
|
},
|
|
*CURATED_STRATEGIES,
|
|
]
|
|
for regime in ("ice", "repair", "fermentation", "climax", "divergence", "retreat"):
|
|
jobs = automatic_screener_jobs(strategies, regime)
|
|
curated_names = {
|
|
job["strategy"]["name"] for job in jobs if job["mode"] == "curated"
|
|
}
|
|
self.assertEqual(
|
|
{strategy["name"] for strategy in CURATED_STRATEGIES},
|
|
curated_names,
|
|
regime,
|
|
)
|
|
|
|
def test_curated_risk_flags_do_not_reintroduce_regime_gating(self):
|
|
row = {
|
|
"pct_chg": 0,
|
|
"return_10d": 0,
|
|
"volatility_10d": 0,
|
|
"amount_billion": 5,
|
|
}
|
|
self.assertIn("市场处于退潮阶段，策略可能选择空仓", _risk_flags(row, "retreat"))
|
|
self.assertNotIn(
|
|
"市场处于退潮阶段，策略可能选择空仓",
|
|
_risk_flags(row, "retreat", include_regime_risk=False),
|
|
)
|
|
|
|
def test_every_curated_formula_uses_supported_factors(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
database = ReviewDatabase(Path(root) / "review.db")
|
|
engine = ScreenerEngine(database)
|
|
for strategy in CURATED_STRATEGIES:
|
|
formula = engine.validate_formula(strategy["formula"])
|
|
fields = {
|
|
item["field"]
|
|
for item in formula["filters"] + formula["score"]
|
|
}
|
|
self.assertTrue(fields.issubset(FACTOR_FIELDS), strategy["name"])
|
|
|
|
def test_factor_groups_cover_every_quant_factor(self):
|
|
grouped = [field for fields in FACTOR_GROUPS.values() for field in fields]
|
|
self.assertEqual(set(FACTOR_FIELDS), set(grouped))
|
|
self.assertEqual(len(grouped), len(set(grouped)))
|
|
|
|
def test_database_migrates_valuation_and_fundamental_columns(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
path = Path(root) / "review.db"
|
|
ReviewDatabase(path)
|
|
connection = sqlite3.connect(path)
|
|
try:
|
|
indicator_columns = {
|
|
row[1] for row in connection.execute("PRAGMA table_info(daily_indicators)")
|
|
}
|
|
tables = {
|
|
row[0] for row in connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'"
|
|
)
|
|
}
|
|
finally:
|
|
connection.close()
|
|
self.assertTrue({"pe_ttm", "pb", "ps_ttm", "dv_ttm"}.issubset(indicator_columns))
|
|
self.assertIn("fundamental_indicators", tables)
|
|
self.assertIn("benchmark_bars", tables)
|
|
|
|
def test_advanced_strategies_declare_history_and_backtest_contracts(self):
|
|
for strategy in ADVANCED_CURATED_STRATEGIES:
|
|
meta = strategy["formula"]["meta"]
|
|
self.assertGreaterEqual(meta["history_days"], 80, strategy["name"])
|
|
self.assertGreaterEqual(meta["backtest_days"], 1, strategy["name"])
|
|
self.assertGreater(meta["take_profit"], 0, strategy["name"])
|
|
self.assertLess(meta["stop_loss"], 0, strategy["name"])
|
|
|
|
def test_quarter_periods_stop_at_selected_date(self):
|
|
periods = _quarter_periods("20260722", 5)
|
|
self.assertEqual(
|
|
["20250630", "20250930", "20251231", "20260331", "20260630"],
|
|
periods,
|
|
)
|
|
|
|
def test_factor_health_summary_uses_availability_counts(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
database = ReviewDatabase(Path(root) / "review.db")
|
|
with database.connect() as connection:
|
|
connection.execute(
|
|
"INSERT INTO daily_bars (trade_date, ts_code) VALUES (?, ?)",
|
|
("20260722", "600000.SH"),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO daily_indicators
|
|
(trade_date, ts_code, pe_ttm)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
("20260722", "600000.SH", 8.5),
|
|
)
|
|
connection.executemany(
|
|
"INSERT INTO daily_indicators (trade_date, ts_code) VALUES (?, ?)",
|
|
[(f"{year}1231", f"{year % 100:02d}0000.SZ") for year in range(2022, 2026)],
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO auction_factors (trade_date, ts_code) VALUES (?, ?)",
|
|
("20260722", "600000.SH"),
|
|
)
|
|
connection.executemany(
|
|
"""
|
|
INSERT INTO fundamental_indicators (end_date, ann_date, ts_code, roe)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
[
|
|
("20251231", "20260430", f"{index:06d}.SZ", 10.0)
|
|
for index in range(100)
|
|
],
|
|
)
|
|
connection.executemany(
|
|
"INSERT INTO benchmark_bars (trade_date, ts_code, close) VALUES (?, ?, ?)",
|
|
[(f"2026{index + 1:04d}", "000300.SH", 4000 + index) for index in range(60)],
|
|
)
|
|
|
|
health = database.factor_health_summary("20260722")
|
|
self.assertTrue(health["market"])
|
|
self.assertTrue(health["auction"])
|
|
self.assertTrue(health["valuation"])
|
|
self.assertTrue(health["fundamental"])
|
|
self.assertTrue(health["dividend_history"])
|
|
self.assertTrue(health["benchmark"])
|
|
self.assertEqual(health["valuation_rows"], 1)
|
|
self.assertEqual(health["fundamental_rows"], 100)
|
|
self.assertEqual(health["dividend_years"], 5)
|
|
|
|
def test_moneyflow_health_requires_the_latest_five_market_dates(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
database = ReviewDatabase(Path(root) / "review.db")
|
|
dates = [f"202607{day:02d}" for day in range(20, 25)]
|
|
database.upsert_daily_bars([
|
|
{
|
|
"trade_date": trade_date, "ts_code": "600000.SH",
|
|
"open": 10, "high": 10.2, "low": 9.8, "close": 10,
|
|
"pct_chg": 0, "vol": 1000, "amount": 100000,
|
|
}
|
|
for trade_date in dates
|
|
])
|
|
database.upsert_moneyflow([
|
|
{"trade_date": "20260105", "ts_code": "600000.SH", "net_mf_amount": 10}
|
|
] * 5)
|
|
self.assertFalse(database.factor_health_summary(dates[-1])["moneyflow_history"])
|
|
database.upsert_moneyflow([
|
|
{"trade_date": trade_date, "ts_code": "600000.SH", "net_mf_amount": 10}
|
|
for trade_date in dates
|
|
])
|
|
health = database.factor_health_summary(dates[-1])
|
|
self.assertTrue(health["moneyflow_history"])
|
|
self.assertEqual(health["moneyflow_dates"], 5)
|
|
|
|
def test_technical_helpers_detect_rsi_and_daily_reversal_path(self):
|
|
self.assertLess(_rsi([10, 9, 8, 7, 6, 5, 4], 6), 1)
|
|
rows = [
|
|
{"close": 10, "high": 10, "vol": 100},
|
|
{"close": 11, "high": 11, "vol": 120},
|
|
{"close": 12, "high": 12, "vol": 130},
|
|
{"close": 11.2, "high": 11.8, "vol": 100},
|
|
{"close": 12.5, "high": 12.5, "vol": 140},
|
|
]
|
|
metrics = _broken_reversal_metrics(
|
|
rows, [False, True, True, False, True], "600000", "示例"
|
|
)
|
|
self.assertEqual(metrics["signal"], 1)
|
|
self.assertEqual(metrics["days"], 1)
|
|
|
|
def test_factor_builder_generates_long_window_and_benchmark_factors(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
database = ReviewDatabase(Path(root) / "review.db")
|
|
database.upsert_stock_master([
|
|
{
|
|
"ts_code": "600000.SH", "name": "趋势样本", "industry": "银行",
|
|
"market": "主板", "list_date": "20000101",
|
|
}
|
|
])
|
|
dates = []
|
|
cursor = datetime(2025, 6, 1)
|
|
while len(dates) < 260:
|
|
if cursor.weekday() < 5:
|
|
dates.append(cursor.strftime("%Y%m%d"))
|
|
cursor += timedelta(days=1)
|
|
bars = []
|
|
benchmarks = []
|
|
indicators = []
|
|
for index, trade_date in enumerate(dates):
|
|
close = 10 + index * 0.05
|
|
bars.append({
|
|
"trade_date": trade_date, "ts_code": "600000.SH",
|
|
"open": close - 0.02, "high": close + 0.08, "low": close - 0.08,
|
|
"close": close, "pct_chg": 0.25, "vol": 1000 + index,
|
|
"amount": 200000,
|
|
})
|
|
benchmarks.append({
|
|
"trade_date": trade_date, "ts_code": "000300.SH",
|
|
"close": 4000 + index, "pct_chg": 0.02,
|
|
})
|
|
if index >= 250:
|
|
indicators.append({
|
|
"trade_date": trade_date, "ts_code": "600000.SH",
|
|
"turnover_rate": 2, "volume_ratio": 1,
|
|
})
|
|
database.upsert_daily_bars(bars)
|
|
database.upsert_benchmark_bars(benchmarks)
|
|
database.upsert_daily_indicators(indicators)
|
|
|
|
factors, actual_date = ScreenerEngine(database).build_factors(
|
|
dates[-1], history_days=260
|
|
)
|
|
|
|
self.assertEqual(actual_date, dates[-1])
|
|
self.assertEqual(len(factors), 1)
|
|
factor = factors[0]
|
|
self.assertEqual(factor["ma_bull_alignment"], 1)
|
|
self.assertEqual(factor["rs_high_120"], 1)
|
|
self.assertGreater(factor["momentum_60_5"], 0)
|
|
self.assertEqual(factor["momentum_60_5_rank"], 0)
|
|
|
|
def test_factor_builder_generates_sector_momentum_and_five_day_flow(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
database = ReviewDatabase(Path(root) / "review.db")
|
|
stocks = [
|
|
("600001.SH", "动量样本", "电子", 0.16, 180),
|
|
("600002.SH", "对照样本", "银行", 0.02, -40),
|
|
]
|
|
database.upsert_stock_master([
|
|
{
|
|
"ts_code": code, "name": name, "industry": industry,
|
|
"market": "主板", "list_date": "20000101",
|
|
}
|
|
for code, name, industry, _, _ in stocks
|
|
])
|
|
dates = []
|
|
cursor = datetime(2026, 4, 1)
|
|
while len(dates) < 80:
|
|
if cursor.weekday() < 5:
|
|
dates.append(cursor.strftime("%Y%m%d"))
|
|
cursor += timedelta(days=1)
|
|
bars = []
|
|
for index, trade_date in enumerate(dates):
|
|
for code, _, _, slope, _ in stocks:
|
|
close = 10 + index * slope
|
|
bars.append({
|
|
"trade_date": trade_date, "ts_code": code,
|
|
"open": close - 0.03, "high": close + 0.08,
|
|
"low": close - 0.08, "close": close,
|
|
"pct_chg": slope, "vol": 1000 + index,
|
|
"amount": 300000,
|
|
})
|
|
database.upsert_daily_bars(bars)
|
|
database.upsert_daily_indicators([
|
|
{
|
|
"trade_date": dates[-1], "ts_code": code,
|
|
"turnover_rate": 2, "volume_ratio": 1,
|
|
"circ_mv": 1000000, "total_mv": 1500000,
|
|
}
|
|
for code, *_ in stocks
|
|
])
|
|
database.upsert_moneyflow([
|
|
{
|
|
"trade_date": trade_date, "ts_code": code,
|
|
"net_mf_amount": daily_flow,
|
|
}
|
|
for trade_date in dates[-5:]
|
|
for code, _, _, _, daily_flow in stocks
|
|
])
|
|
|
|
factors, _ = ScreenerEngine(database).build_factors(
|
|
dates[-1], history_days=80
|
|
)
|
|
by_code = {item["ts_code"]: item for item in factors}
|
|
leader = by_code["600001.SH"]
|
|
laggard = by_code["600002.SH"]
|
|
self.assertGreater(leader["return_20d"], laggard["return_20d"])
|
|
self.assertEqual(leader["sector_momentum_rank"], 1)
|
|
self.assertEqual(laggard["sector_momentum_rank"], 0)
|
|
self.assertGreater(leader["net_flow_5d_million"], 0)
|
|
self.assertLess(laggard["net_flow_5d_million"], 0)
|
|
self.assertEqual(leader["sector_flow_rank"], 1)
|
|
|
|
def test_screen_reports_signal_health(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
database = ReviewDatabase(Path(root) / "review.db")
|
|
engine = ScreenerEngine(database)
|
|
formula = {
|
|
"universe": {"exclude_st": True, "listed_days_min": 0},
|
|
"filters": [{"field": "pct_chg", "op": ">", "value": 0}],
|
|
"score": [{"field": "amount_billion", "weight": 1, "direction": "desc"}],
|
|
"limit": 5,
|
|
"min_score": 0,
|
|
}
|
|
result = engine.screen(
|
|
0, "20260724", formula, "repair", "健康检查", False,
|
|
mode="curated",
|
|
prepared_factors=[{
|
|
"ts_code": "600000.SH", "code": "600000", "name": "浦发银行",
|
|
"sector": "银行", "listed_days": 1000, "pct_chg": 1,
|
|
"amount_billion": 5, "price": 10, "return_5d": 1,
|
|
"volume_ratio_5d": 1, "sector_strength": 50,
|
|
}],
|
|
prepared_date="20260724",
|
|
)
|
|
health = result["meta"]["health"]
|
|
self.assertEqual(health["status"], "normal")
|
|
self.assertEqual(health["signal_count"], 1)
|
|
self.assertEqual(health["coverage"], 100)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|