aszerW
9e1ba2db03
- strategies/screener/ → archive/legacy_screener/ - base.py (Screener抽象基类) 功能已被 framework/signals/SignalGenerator 覆盖 - cci.py (CCI策略) 依赖已归档的 core.factors.technical 和 core.common 保留新框架结构，使用 SignalGenerator 替代 Screener 抽象
69 lines
1.7 KiB
Python
69 lines
1.7 KiB
Python
"""
|
||
标的筛选器基类
|
||
|
||
用于基于技术指标筛选符合条件的标的
|
||
"""
|
||
|
||
from abc import ABC, abstractmethod
|
||
from typing import Any
|
||
import pandas as pd
|
||
|
||
|
||
class Screener(ABC):
|
||
"""筛选器抽象基类"""
|
||
|
||
def __init__(self, name: str, config: dict = None):
|
||
self.name = name
|
||
self.config = config or {}
|
||
|
||
@abstractmethod
|
||
def screen(self, data: Any) -> dict:
|
||
"""
|
||
执行筛选
|
||
|
||
Args:
|
||
data: 输入数据（DataFrame或其他格式）
|
||
|
||
Returns:
|
||
dict: 筛选结果，必须包含 'triggered' 键表示是否触发
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def screen_batch(self, data_dict: dict) -> list:
|
||
"""
|
||
批量筛选多个标的
|
||
|
||
Args:
|
||
data_dict: {code: data} 格式的字典
|
||
|
||
Returns:
|
||
list: 符合条件的标的列表
|
||
"""
|
||
pass
|
||
|
||
|
||
class DataFrameScreener(Screener):
|
||
"""基于DataFrame的筛选器基类"""
|
||
|
||
def __init__(self, name: str, config: dict = None):
|
||
super().__init__(name, config)
|
||
|
||
def validate_data(self, df: pd.DataFrame) -> bool:
|
||
"""验证数据格式"""
|
||
required_cols = ["open", "high", "low", "close", "volume"]
|
||
return all(col in df.columns for col in required_cols)
|
||
|
||
def screen_batch(self, data_dict: dict) -> list:
|
||
"""批量筛选"""
|
||
results = []
|
||
for code, data in data_dict.items():
|
||
if isinstance(data, pd.DataFrame) and self.validate_data(data):
|
||
result = self.screen(data)
|
||
if result.get("triggered", False):
|
||
results.append({
|
||
"code": code,
|
||
**result
|
||
})
|
||
return results
|