53 lines
2.0 KiB
Python
53 lines
2.0 KiB
Python
# -*- 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
|