社区分享｜QClaw × PandaAI 全自动因子挖掘实战

# QClaw随OpenClaw自动安装# 无需额外配置# 验证安装openclaw --version# 输出: QClaw v1.0.0

# 克隆仓库git clone https://github.com/yourusername/pandafactor.gitcd pandafactor# 安装依赖pip install -r requirements.txt# 配置MongoDB（可选）# 编辑 config.yamlmongo:  host: localhost  port: 27017  database: pandafactor

# 无需额外操作，QClaw自动生成# 见后文"因子挖掘流程"

# 使用Wind数据源from WindPy import ww.start()# 使用Tushare数据源import tushare as tsts.set_token('your_token')pro = ts.pro_api()# 使用聚宽数据源import jqdatasdk as jqjq.auth('your_phone', 'your_password')

# QClaw自动执行（用户无感知）node prosearch.cjs --keyword="量化策略研报 因子投资" --freshness=30d# 找到5家机构研报：# 1. 华宝证券：动量风格强势# 2. 万家科技：高频量价策略# 3. 中银国际：低换手率因子变强# 4. 鹏华基金：基本面量化因子库# 5. 源达信息：重要股东增持因子

# QClaw自动生成的因子列表（factor_mining_top10.py）class FactorLibrary:    def momentum_10d(self):        """10日动量因子（华宝证券思路）"""        return self.close / self.close.shift(10) - 1    def macd_histogram(self):        """MACD柱因子（万家科技思路）"""        ema12 = self.close.ewm(span=12).mean()        ema26 = self.close.ewm(span=26).mean()        dif = ema12 - ema26        dea = dif.ewm(span=9).mean()        return (dif - dea) * 2    def turnover_5d(self):        """5日换手率因子（中银国际思路）"""        return self.volume.rolling(5).sum() / self.close    # ... 共25个因子

# QClaw自动调用PandaAI（用户无感知）def backtest_factors(ohlcv, factors_dict, holding_period=5):    """批量回测因子"""    results = []    future_returns = ohlcv['close'].shift(-holding_period) / ohlcv['close'] - 1    for factor_name, factor_values in factors_dict.items():        # 计算IC        ic = factor_values.corr(future_returns)        # 计算Rank IC        rank_ic = factor_values.rank().corr(future_returns.rank())        # 分组回测        factor_df['factor_rank'] = factor_values.rank(pct=True)        long_returns = factor_df[factor_df['factor_rank'] >= 0.8]['returns'].mean()        short_returns = factor_df[factor_df['factor_rank'] <= 0.2]['returns'].mean()        results.append({            'factor_name': factor_name,            'ic': ic,            'rank_ic': rank_ic,            'long_returns': long_returns,            'short_returns': short_returns,            'long_short_returns': long_returns - short_returns        })    return pd.DataFrame(results)

# QClaw自动筛选有效因子valid_factors = results_df[abs(results_df['ic']) > 0.02]top10 = valid_factors.nlargest(10, 'rank_ic')# 输出完整代码文件（top10_valid_factors.py）# 包含：因子计算函数、详细注释、使用示例、标准化函数

# top10_valid_factors.py (19.8 KB)# 包含10个因子的完整实现def calc_macd_histogram(close: pd.Series) -> pd.Series:    """    MACD柱因子    研报依据：万家科技高频量价类策略    回测表现：IC=0.0647, Rank IC=0.0922    使用示例：        close = pd.Series([...])        macd = calc_macd_histogram(close)        signal = macd.apply(lambda x: 1 if x > 0 else -1)    """    ema12 = close.ewm(span=12, adjust=False).mean()    ema26 = close.ewm(span=26, adjust=False).mean()    dif = ema12 - ema26    dea = dif.ewm(span=9, adjust=False).mean()    return (dif - dea) * 2

def main():    # [1/4] 生成模拟市场数据    ohlcv = generate_market_data('2018-01-01', '2025-12-31')    # [2/4] 计算因子（基于研报思路）    factor_lib = FactorLibrary(ohlcv)    factors_dict = factor_lib.calculate_all_factors()  # 25个因子    # [3/4] 批量回测因子表现    results_df = backtest_factors(ohlcv, factors_dict, holding_period=5)    # [4/4] 筛选有效因子（IC > 0.02）    valid_factors = results_df[abs(results_df['ic']) > 0.02]    top10 = valid_factors.nlargest(10, 'rank_ic')    return top10

# ❌ 传统方法：逐个计算for factor_name in factor_names:    factor_values[factor_name] = calc_factor(factor_name)# 耗时：约30秒# ✅ 优化方法1：向量化计算all_factors = pd.DataFrame()for factor_name in factor_names:    all_factors[factor_name] = calc_factor(factor_name)# 耗时：约5秒# ✅ 优化方法2：并行计算from concurrent.futures import ThreadPoolExecutorwith ThreadPoolExecutor(max_workers=4) as executor:    futures = [executor.submit(calc_factor, f) for f in factor_names]    results = [f.result() for f in futures]# 耗时：约2秒

class FactorBacktest:    """因子回测框架"""    def __init__(self, ohlcv, holding_period=5, groups=5):        self.ohlcv = ohlcv        self.holding_period = holding_period        self.groups = groups    def calculate_ic(self, factor_values):        """计算IC值"""        future_returns = self.ohlcv['close'].shift(-self.holding_period) / self.ohlcv['close'] - 1        return factor_values.corr(future_returns)    def group_backtest(self, factor_values):        """分组回测"""        factor_df = pd.DataFrame({            'factor': factor_values,            'returns': self.future_returns        }).dropna()        # 分组        factor_df['group'] = pd.qcut(factor_df['factor'], self.groups, labels=False)        # 计算每组收益        group_returns = factor_df.groupby('group')['returns'].mean()        return {            'long': group_returns[self.groups - 1],  # 最高组            'short': group_returns[0],  # 最低组            'long_short': group_rewards[self.groups - 1] - group_rewards[0]        }

# ❌ 传统方法：CSV文件df = pd.read_csv('stock_data.csv')  # 耗时：30秒# ✅ 优化方法1：Parquet格式df = pd.read_parquet('stock_data.parquet')  # 耗时：2秒# ✅ 优化方法2：MongoDB（PandaAI内置）from pymongo import MongoClientclient = MongoClient('localhost', 27017)db = client['pandafactor']collection = db['stock_daily']df = pd.DataFrame(list(collection.find()))  # 耗时：1秒

# 使用multiprocessing并行计算import multiprocessing as mpdef parallel_factor_calculation(factor_func, ohlcv_chunk):    """并行计算因子"""    return factor_func(ohlcv_chunk)if __name__ == '__main__':    # 分块    chunks = np.array_split(ohlcv, mp.cpu_count())    # 并行计算    with mp.Pool(mp.cpu_count()) as pool:        results = pool.starmap(            parallel_factor_calculation,            [(calc_momentum_10d, chunk) for chunk in chunks]        )    # 合并结果    factor_values = pd.concat(results)

# ❌ 占用大量内存all_factors = pd.DataFrame()for i in range(1000):    all_factors[f'factor_{i}'] = calc_factor(i)# 内存占用：约10GB# ✅ 使用迭代器节省内存def factor_generator(factor_names):    for name in factor_names:        yield name, calc_factor(name)# 逐个处理，不一次性加载for name, values in factor_generator(factor_names):    ic = calculate_ic(values)    store_result(name, ic)# 内存占用：约500MB

ServerSelectionTimeoutError: localhost:27017: [Errno 111] Connection refused

# 1. 检查MongoDB是否运行sudo systemctl status mongod# 2. 启动MongoDBsudo systemctl start mongod# 3. 或者使用独立模式（无需MongoDB）python factor_mining_top10.py --no-mongo

TimeoutError: Factor calculation timeout after 60 seconds

# 1. 减少因子数量factor_names = ['momentum_10d', 'macd_histogram', 'rsi_14']  # 只计算3个# 2. 缩小数据范围ohlcv = generate_market_data('2020-01-01', '2025-12-31')  # 只用5年数据# 3. 并行计算# 见"性能优化技巧"

未发现有效因子（IC > 0.02）

# 1. 降低筛选阈值valid_factors = results_df[abs(results_df['ic']) > 0.01]  # IC > 0.01# 2. 检查数据质量print(ohlcv.isnull().sum())  # 检查缺失值print(ohlcv.describe())  # 检查异常值# 3. 增加因子数量factor_names = [f for f in dir(FactorLibrary) if not f.startswith('_')]# 从25个增加到50个

UnicodeEncodeError: 'gbk' codec can't encode character '\u2713'

# 在代码开头添加import sysimport iosys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')# 或使用 qclaw-text-file 技能写入文件# 自动处理编码问题

# 1. 安装环境pip install pandafactor# 2. 运行示例python factor_mining_top10.py# 3. 查看结果cat valid_factors_top10.csv

# 1. 连接真实数据源修改 config.yaml 中的数据源配置# 2. 扩展因子库在 FactorLibrary 类中添加新因子# 3. 优化回测框架调整持仓周期、分组数量等参数

# 1. 构建因子组合使用机器学习优化因子权重# 2. 实盘部署接入券商API，自动化交易# 3. 持续迭代定期更新因子，动态调整权重

# 1. 使用版本管理git initgit add factor_mining_top10.pygit commit -m "初始因子挖掘脚本"# 2. 定期备份数据cp stock_data.parquet stock_data_backup_20260424.parquet# 3. 记录实验日志

# 1. 不要直接修改核心代码# ❌ 修改 FactorLibrary 基类# ✅ 继承并扩展class CustomFactorLibrary(FactorLibrary):    def my_new_factor(self):        return ...# 2. 不要忽略警告import warningswarnings.filterwarnings('ignore')  # ❌ 隐藏所有警告warnings.filterwarnings('ignore', category=FutureWarning)  # ✅ 只隐藏特定警告# 3. 不要硬编码路径# ❌ df.to_csv('C:\\Users\\...\\data.csv')# ✅ df.to_csv(f'{workspace_root}/data.csv')

.qclaw/workspace/├── data/                        # 数据目录│   ├── stock_data.parquet      # 原始数据│   └── factor_values.parquet   # 因子值数据│├── factors/                     # 因子代码目录│   ├── factor_mining_top10.py  # 因子挖掘脚本│   ├── top10_valid_factors.py  # Top 10因子代码│   └── custom_factors.py       # 自定义因子│├── backtest/                    # 回测结果目录│   ├── all_factors_backtest.csv│   ├── valid_factors_top10.csv│   └── backtest_reports/       # 回测报告│├── strategies/                  # 策略代码目录│   ├── insider_buying_strategy.py│   └── momentum_strategy.py│└── docs/                        # 文档目录    ├── task-summary_20260423-factor-mining-top10.md    └── qclaw_pandaAI_auto_factor_mining_guide.md

# factor_mining_simple.pyimport pandas as pdimport numpy as npdef generate_data():    """生成模拟数据"""    dates = pd.date_range('2020-01-01', '2025-12-31', freq='B')    close = 100 * np.exp(np.cumsum(np.random.randn(len(dates)) * 0.02))    return pd.DataFrame({'close': close}, index=dates)def calc_momentum_10d(close):    """10日动量因子"""    return close / close.shift(10) - 1def calc_macd(close):    """MACD柱因子"""    ema12 = close.ewm(span=12).mean()    ema26 = close.ewm(span=26).mean()    dif = ema12 - ema26    dea = dif.ewm(span=9).mean()    return (dif - dea) * 2def main():    # 生成数据    ohlcv = generate_data()    # 计算因子    factors = {        'momentum_10d': calc_momentum_10d(ohlcv['close']),        'macd': calc_macd(ohlcv['close'])    }    # 回测    future_returns = ohlcv['close'].shift(-5) / ohlcv['close'] - 1    for name, values in factors.items():        ic = values.corr(future_returns)        print(f"{name}: IC={ic:.4f}")    return factorsif __name__ == '__main__':    main()

# strategy_framework_simple.pyimport pandas as pdclass SimpleFactorStrategy:    """简单因子策略框架"""    def __init__(self, factor_func, top_n=10):        self.factor_func = factor_func        self.top_n = top_n    def generate_signal(self, ohlcv):        """生成交易信号"""        factor_values = self.factor_func(ohlcv)        # 排序        factor_rank = factor_values.rank()        # 买入前N只        long_stocks = factor_rank.nlargest(self.top_n).index        return {            'long': long_stocks,            'signal': 1  # 买入信号        }# 使用示例def momentum_factor(ohlcv):    return ohlcv['close'] / ohlcv['close'].shift(10) - 1strategy = SimpleFactorStrategy(momentum_factor, top_n=10)signal = strategy.generate_signal(ohlcv)

# 一键安装pip install pandafactor# 运行示例python factor_mining_top10.py# 获得Top 10有效因子cat valid_factors_top10.csv