Python中如何实现一个Pythonic的i问财量化策略工具包

要做一个Pythonic的i问财策略工具包，核心是遵循Python的设计哲学（简洁、可读、明确）。下面是一个高度抽象但完整的框架示例，展示了关键组件的实现思路：

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Optional, Callable
import pandas as pd
import numpy as np

@dataclass
class Stock:
    """股票数据类"""
    code: str
    name: str
    data: pd.DataFrame  # 包含日期、开盘、收盘等字段

class DataFetcher:
    """数据获取器 - 负责从i问财获取数据"""
    
    @staticmethod
    def fetch_stock_data(query: str, start_date: str, end_date: str) -> List[Stock]:
        """
        执行i问财查询并返回股票列表
        
        Args:
            query: i问财查询语句，如'pe<20 and roe>15'
            start_date: 开始日期 '2023-01-01'
            end_date: 结束日期 '2023-12-31'
        """
        # 这里应该实现实际的i问财API调用
        # 示例返回模拟数据
        stocks = []
        # 模拟查询结果
        results = ['000001.SZ', '000002.SZ']  # 假设查询返回这些股票
        
        for code in results:
            # 获取历史数据（这里需要实际实现）
            data = pd.DataFrame({
                'date': pd.date_range(start_date, end_date),
                'close': np.random.randn(100).cumsum() + 100
            })
            stocks.append(Stock(code=code, name=f"股票{code}", data=data))
        
        return stocks

class Strategy(ABC):
    """策略基类 - 定义策略接口"""
    
    @abstractmethod
    def filter(self, stock: Stock) -> bool:
        """过滤股票条件"""
        pass
    
    @abstractmethod
    def signal(self, stock: Stock) -> Optional[float]:
        """生成交易信号"""
        pass

class MeanReversionStrategy(Strategy):
    """均值回归策略示例"""
    
    def __init__(self, window: int = 20, threshold: float = 2.0):
        self.window = window
        self.threshold = threshold
    
    def filter(self, stock: Stock) -> bool:
        """过滤条件：至少有window+10个数据点"""
        return len(stock.data) > self.window + 10
    
    def signal(self, stock: Stock) -> Optional[float]:
        """生成信号：当价格偏离移动平均线超过阈值时发出信号"""
        closes = stock.data['close'].values
        if len(closes) < self.window:
            return None
        
        ma = np.mean(closes[-self.window:])
        current = closes[-1]
        z_score = (current - ma) / np.std(closes[-self.window:])
        
        if z_score < -self.threshold:
            return 1.0  # 买入信号
        elif z_score > self.threshold:
            return -1.0  # 卖出信号
        return 0.0  # 持有

class Backtester:
    """回测引擎"""
    
    def __init__(self, strategy: Strategy):
        self.strategy = strategy
    
    def run(self, stocks: List[Stock]) -> pd.DataFrame:
        """运行回测"""
        results = []
        
        for stock in stocks:
            if not self.strategy.filter(stock):
                continue
            
            signal = self.strategy.signal(stock)
            if signal is not None:
                results.append({
                    'code': stock.code,
                    'name': stock.name,
                    'signal': signal,
                    'last_close': stock.data['close'].iloc[-1]
                })
        
        return pd.DataFrame(results)

# 使用示例
def main():
    # 1. 获取数据
    fetcher = DataFetcher()
    stocks = fetcher.fetch_stock_data('pe<20 and roe>15', '2023-01-01', '2023-12-31')
    
    # 2. 创建策略
    strategy = MeanReversionStrategy(window=30, threshold=1.5)
    
    # 3. 回测
    backtester = Backtester(strategy)
    results = backtester.run(stocks)
    
    # 4. 输出结果
    print("策略选股结果:")
    print(results.to_string(index=False))

if __name__ == "__main__":
    main()

关键设计要点：

1. 面向对象设计：使用抽象基类定义接口，具体策略继承实现
2. 数据类：用@dataclass简化数据容器
3. 类型提示：全程使用type hints提高代码可读性
4. 单一职责：每个类只负责一个明确的功能
5. 函数式风格：策略的filter和signal方法应该是纯函数

扩展建议：

• 添加装饰器实现策略的性能统计
• 使用__call__方法让策略对象可调用
• 实现策略组合和仓位管理模块

这个框架遵循了Python的"EAFP"（Easier to Ask for Forgiveness than Permission）原则，通过清晰的接口设计和合理的抽象层次，让策略开发变得直观且易于维护。

用dataclass和抽象基类来组织代码结构最Pythonic。