refactor: 移除screener目录到archive

- strategies/screener/ → archive/legacy_screener/
- base.py (Screener抽象基类) 功能已被 framework/signals/SignalGenerator 覆盖
- cci.py (CCI策略) 依赖已归档的 core.factors.technical 和 core.common

保留新框架结构，使用 SignalGenerator 替代 Screener 抽象

archive/legacy_screener/base.py

@@ -0,0 +1,68 @@
+"""
+标的筛选器基类
+
+用于基于技术指标筛选符合条件的标的
+"""
+
+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

archive/legacy_screener/cci.py

@@ -0,0 +1,186 @@
+"""
+CCI技术指标筛选器
+
+基于商品通道指数(CCI)筛选超卖标的
+"""
+
+import pandas as pd
+from datetime import datetime
+
+from .base import DataFrameScreener
+from core.factors.technical import calculate_cci, resample_to_weekly
+from core.common.db import DatabaseManager
+from core.common.notify import NotificationManager
+
+
+class CCIScreener(DataFrameScreener):
+    """CCI超卖筛选器"""
+
+    def __init__(self, config: dict = None):
+        super().__init__("CCI超卖筛选", config)
+        self.day_period = config.get("day_period", 14)
+        self.week_period = config.get("week_period", 14)
+        self.threshold = config.get("threshold", -100)
+        self.use_weekly = config.get("use_weekly", True)
+        self.db_manager = DatabaseManager()
+        self.notifier = NotificationManager()
+
+    def screen(self, df: pd.DataFrame) -> dict:
+        """
+        对单只标的进行CCI筛选
+
+        Args:
+            df: DataFrame with OHLCV data
+
+        Returns:
+            dict: {
+                'triggered': bool,
+                'day_cci': float,
+                'week_cci': float or None,
+                'current_price': float,
+            }
+        """
+        if not self.validate_data(df):
+            return {"triggered": False, "error": "数据格式错误"}
+
+        # 计算日线CCI
+        day_cci = calculate_cci(df, period=self.day_period).iloc[-1]
+        current_price = df["close"].iloc[-1]
+
+        result = {
+            "triggered": day_cci < self.threshold,
+            "day_cci": round(day_cci, 2),
+            "week_cci": None,
+            "current_price": round(current_price, 2),
+        }
+
+        # 计算周线CCI
+        if self.use_weekly:
+            weekly_df = resample_to_weekly(df)
+            if len(weekly_df) >= self.week_period:
+                week_cci = calculate_cci(weekly_df, period=self.week_period).iloc[-1]
+                result["week_cci"] = round(week_cci, 2)
+                # 日线或周线任一超卖即触发
+                result["triggered"] = (
+                    day_cci < self.threshold or week_cci < self.threshold
+                )
+
+        return result
+
+    def get_data_from_db(self, code: str, limit: int = 100) -> pd.DataFrame:
+        """从数据库获取数据"""
+        sql = f"""
+            SELECT date, open, high, low, close, volume
+            FROM public.index_kline
+            WHERE code = '{code}'
+            ORDER BY date DESC
+            LIMIT {limit}
+        """
+        result = self.db_manager.execute_query(sql)
+
+        if not result:
+            return pd.DataFrame()
+
+        df = pd.DataFrame(result)
+        df["date"] = pd.to_datetime(df["date"])
+
+        # 转换数值类型
+        for col in ["open", "high", "low", "close", "volume"]:
+            df[col] = pd.to_numeric(df[col], errors="coerce")
+
+        df = df.sort_values("date").reset_index(drop=True)
+        return df
+
+    def run_screening(self, code_list: list = None) -> list:
+        """
+        执行批量筛选
+
+        Args:
+            code_list: 标的代码列表，None则从配置文件读取
+
+        Returns:
+            list: 符合条件的标的列表
+        """
+        if code_list is None:
+            # 从CSV文件读取指数列表
+            import os
+            csv_path = self.config.get("index_fund_info_file", "index_fund_info.csv")
+            if os.path.exists(csv_path):
+                df = pd.read_csv(csv_path, encoding="utf-8-sig")
+                code_list = df.drop_duplicates(subset=["指数代码"]).to_dict("records")
+            else:
+                raise ValueError(f"找不到标的列表文件: {csv_path}")
+
+        signals = []
+        today_str = datetime.now().strftime("%Y-%m-%d")
+
+        print(f"开始CCI筛选，共 {len(code_list)} 个标的...")
+
+        for i, code_info in enumerate(code_list):
+            if isinstance(code_info, dict):
+                code = code_info.get("指数代码")
+                name = code_info.get("指数名称", code)
+            else:
+                code = code_info
+                name = code
+
+            try:
+                df = self.get_data_from_db(code, limit=self.config.get("lookback_days", 100))
+
+                if len(df) < self.day_period:
+                    continue
+
+                # 检查最新日期
+                if df["date"].max().strftime("%Y-%m-%d") != today_str:
+                    continue
+
+                result = self.screen(df)
+
+                if result["triggered"]:
+                    signals.append({
+                        "code": code,
+                        "name": name,
+                        "day_cci": result["day_cci"],
+                        "week_cci": result["week_cci"],
+                        "price": result["current_price"],
+                    })
+                    print(f"  ✓ {code} ({name}): 日线CCI={result['day_cci']:.2f}")
+
+            except Exception as e:
+                print(f"  ✗ {code}: {e}")
+                continue
+
+        print(f"\n筛选完成，{len(signals)} 个标的符合CCI超卖条件")
+
+        # 发送通知
+        if signals:
+            self.notifier.notify_signal(signals, signal_type="CCI超卖")
+
+        return signals
+
+    def run_daily(self):
+        """每日定时运行"""
+        from datetime import datetime
+
+        # 检查是否为交易日
+        if datetime.today().weekday() >= 5 and self.config.get("skip_weekend", True):
+            print("非交易日，跳过")
+            return
+
+        self.run_screening()
+
+
+def create_cci_screener_from_config(config_path: str = None) -> CCIScreener:
+    """从配置文件创建CCI筛选器"""
+    import yaml
+    import os
+
+    if config_path is None:
+        config_path = os.path.join(
+            os.path.dirname(__file__), "..", "..", "config", "strategies", "cci.yaml"
+        )
+
+    with open(config_path, "r", encoding="utf-8") as f:
+        config = yaml.safe_load(f)
+
+    return CCIScreener(config)