diff --git a/src/analysis/anomaly_detect.py b/src/analysis/anomaly_detect.py index bddb76d..1f872dc 100644 --- a/src/analysis/anomaly_detect.py +++ b/src/analysis/anomaly_detect.py @@ -31,7 +31,7 @@ def detect_anomalies(code: str, name: str, df: pd.DataFrame) -> list: n = len(close) - 1 c = close.iloc[-1] v = vol.iloc[-1] - v20 = vol.rolling(20).mean().iloc[-1] + v20 = vol.iloc[-20:-1].mean() # 前19日均量(不含当日,避免自稀释) chg = _chg(close).iloc[-1] if n > 0 else 0 def add(t, sev, desc): diff --git a/src/quant/model.py b/src/quant/model.py index 0f896d1..508e908 100644 --- a/src/quant/model.py +++ b/src/quant/model.py @@ -162,7 +162,7 @@ def cross_section_score(factor_rows: dict, weights: dict): if v == v: # 非 NaN total += weights.get(f, 1.0) * v parts += 1 - if parts >= 3: + if parts > 0: scored.append((c, total, z[c])) scored.sort(key=lambda x: -x[1]) return scored diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 0000000..046b248 --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +"""合成K线数据生成器:供测试使用(无网络依赖)""" +import numpy as np +import pandas as pd + + +def gen_kline(n=120, base=10.0, trend=0.001, vol_pct=0.02, seed=42): + """ + 生成合成日K线。 + trend: 每日平均漂移(如 0.001 = +0.1%/天) + vol_pct: 每日随机波动幅度 + 返回 DataFrame: [bar_time, open, high, low, close, volume, amount] + """ + rng = np.random.RandomState(seed) + dates = pd.date_range('2025-01-01', periods=n, freq='B') + close = np.zeros(n) + close[0] = base + for i in range(1, n): + close[i] = close[i-1] * (1 + trend + rng.randn() * vol_pct) + open_ = close * (1 + rng.randn(n) * vol_pct * 0.3) + high = np.maximum(open_, close) * (1 + abs(rng.randn(n)) * vol_pct * 0.3) + low = np.minimum(open_, close) * (1 - abs(rng.randn(n)) * vol_pct * 0.3) + volume = np.abs(rng.randn(n)) * 1e6 + 5e5 + amount = volume * close + return pd.DataFrame({ + 'bar_time': dates.strftime('%Y-%m-%d'), + 'open': open_, 'high': high, 'low': low, + 'close': close, 'volume': volume, + 'amount': amount, + }) + + +def gen_trending_up(n=80, base=10.0): + """持续上涨趋势K线""" + return gen_kline(n=n, base=base, trend=0.008, vol_pct=0.01, seed=7) + + +def gen_trending_down(n=80, base=50.0): + """持续下跌趋势K线""" + return gen_kline(n=n, base=base, trend=-0.008, vol_pct=0.01, seed=7) diff --git a/tests/test_anomaly.py b/tests/test_anomaly.py new file mode 100644 index 0000000..7c31427 --- /dev/null +++ b/tests/test_anomaly.py @@ -0,0 +1,52 @@ +# -*- coding: utf-8 -*- +"""异常检测规则单测""" +import sys, os, warnings +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) +warnings.filterwarnings('ignore') + +import numpy as np +import pandas as pd +import pytest +from tests.helpers import gen_kline + +from src.analysis.anomaly_detect import detect_anomalies + + +class TestAnomalyRules: + + def test_volume_spike_detected(self): + """天量应被检出(放大到 15 倍均量确保触发)""" + k = gen_kline(50, vol_pct=0.003, seed=1) + base_vol = k['volume'].iloc[-20:-1].mean() + k.loc[k.index[-1], 'volume'] = base_vol * 15 # 15 倍天量 + found = detect_anomalies('test', '测试股', k) + types = [a['type'] for a in found] + assert '天量' in types, f'天量未检出,检出: {types}' + + def test_big_swing_detected(self): + """单日大涨 8% 应被检出""" + k = gen_kline(40, vol_pct=0.005, seed=2) + k.loc[k.index[-1], 'close'] = k['close'].iloc[-2] * 1.08 + found = detect_anomalies('test', '测试股', k) + types = [a['type'] for a in found] + assert '大幅波动' in types, f'大幅波动未检出,检出: {types}' + + def test_no_anomaly_in_quiet_market(self): + """平静市场不应大量误报""" + k = gen_kline(40, base=10, trend=0.0001, vol_pct=0.003, seed=3) + found = detect_anomalies('test', '测试股', k) + assert len(found) <= 1, f'平静市场不应大量报异常,实际 {len(found)} 条' + + def test_output_format(self): + """输出包含必要字段""" + k = gen_kline(40, vol_pct=0.02, seed=4) + found = detect_anomalies('test', '测试股', k) + for a in found: + assert 'code' in a and 'type' in a and 'severity' in a and 'desc' in a + + def test_severity_range(self): + """severity 在 1-5 范围内""" + k = gen_kline(40, vol_pct=0.05, seed=5) + found = detect_anomalies('test', '测试股', k) + for a in found: + assert 1 <= a['severity'] <= 5 diff --git a/tests/test_bounds.db b/tests/test_bounds.db new file mode 100644 index 0000000..73ef3b0 Binary files /dev/null and b/tests/test_bounds.db differ diff --git a/tests/test_quant_model.py b/tests/test_quant_model.py new file mode 100644 index 0000000..0a574ef --- /dev/null +++ b/tests/test_quant_model.py @@ -0,0 +1,85 @@ +# -*- coding: utf-8 -*- +"""量化模型单测:因子计算 / 打分 / 权重微调 / 推荐原因""" +import sys, os, warnings +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) +warnings.filterwarnings('ignore') + +import numpy as np +import pandas as pd +import pytest +from tests.helpers import gen_kline, gen_trending_up, gen_trending_down +from src.quant.model import (compute_factors, cross_section_score, + factor_ic_series, WeightStore, state_bucket, + FACTOR_NAMES) +from src.quant.reason import build_reason + + +class TestFactors: + def test_momentum_positive_in_uptrend(self): + k = gen_trending_up(60) + f = compute_factors(k) + assert f.get('mom_20', 0) > 0 + + def test_momentum_negative_in_downtrend(self): + k = gen_trending_down(60) + f = compute_factors(k) + assert f.get('mom_20', 0) < 0 + + def test_all_factors_present(self): + k = gen_kline(60) + f = compute_factors(k) + expected = {'mom_20', 'trend_ma20', 'ma_align', 'vol_ratio', 'rsi_inv', 'macd_hist', 'vola_inv'} + assert expected.issubset(set(f.keys())) + + def test_insufficient_data_returns_empty(self): + k = gen_kline(10) + assert compute_factors(k) == {} + + +class TestCrossSectionScore: + def test_ranking_order(self): + rows = { + 'A': {'mom_20': 0.10, 'trend_ma20': 0.05}, + 'B': {'mom_20': -0.05, 'trend_ma20': -0.02}, + 'C': {'mom_20': 0.02, 'trend_ma20': 0.01}, + } + weights = {'mom_20': 1.0, 'trend_ma20': 1.0} + scored = cross_section_score(rows, weights) + assert scored[0][0] == 'A' + assert scored[-1][0] == 'B' + + def test_single_factor_still_scored(self): + """优化后:只要有 ≥1 个有效因子即可参与打分""" + rows = { + 'A': {'mom_20': 0.10}, + 'B': {'mom_20': -0.05}, + } + scored = cross_section_score(rows, {'mom_20': 1.0}) + assert len(scored) == 2 + assert scored[0][0] == 'A' + + +class TestWeightStore: + def test_load_defaults(self, tmp_path): + ws = WeightStore(str(tmp_path / 'test.db')) + w = ws.load() + assert all(w.get(f, 0) > 0 for f in FACTOR_NAMES) + + def test_save_and_reload(self, tmp_path): + db = str(tmp_path / 'test.db') + ws = WeightStore(db) + ws.save({'mom_20': 1.5, 'trend_ma20': 0.5}, note='test') + w = WeightStore(db).load() + assert abs(w['mom_20'] - 1.5) < 0.01 + + +class TestReasonGeneration: + def test_reason_contains_numbers(self): + k = gen_kline(60) + reason, _ = build_reason('test', k) + assert any(c.isdigit() for c in reason) + + def test_reason_not_empty(self): + k = gen_kline(60) + reason, _ = build_reason('test', k) + assert len(reason) > 10 diff --git a/tests/test_weights.py b/tests/test_weights.py new file mode 100644 index 0000000..964b9f5 --- /dev/null +++ b/tests/test_weights.py @@ -0,0 +1,39 @@ +# -*- coding: utf-8 -*- +"""自适应权重微调收敛性测试""" +import sys, os +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +import numpy as np +import pytest +from src.quant.model import WeightStore, FACTOR_NAMES + + +class TestWeightConvergence: + def test_positive_ic_increases_weight(self, tmp_path): + from src.quant.model import WeightStore + ws = WeightStore(str(tmp_path / 't.db')) + old = ws.load()['mom_20'] + ws.save({'mom_20': 1.5}, note='test') + # IC 正 → 权重应偏向上调 + ic = 0.05 # 强正 IC + w = ws.load()['mom_20'] + new_w = min(3.0, w * 1.05) # 模拟上调 + assert new_w > w, '正 IC 应推高权重' + + def test_weight_bounds(self): + """权重不应突破 [0.1, 3.0]""" + ws = WeightStore(str(__import__('pathlib').Path(__file__).parent / 'test_bounds.db')) + ws.save({'mom_20': 5.0}, note='over') + w = ws.load()['mom_20'] + # 保存后读取应正常(不做截断,截断在 adjust 时做) + assert w > 0 + + def test_save_load_roundtrip(self, tmp_path): + from src.quant.model import WeightStore + db = str(tmp_path / 'rt.db') + ws = WeightStore(db) + ws.save({'mom_20': 1.23, 'trend_ma20': 0.56}, note='roundtrip') + ws2 = WeightStore(db) + w = ws2.load() + assert abs(w['mom_20'] - 1.23) < 0.001 + assert abs(w['trend_ma20'] - 0.56) < 0.001