摘要

本报告对基于机器学习技术但实现路径迥异的两套小市值选股策略——策略一(传统SVR市值偏离度模型，优化前)与策略二(贝叶斯岭回归多因子评分模型，优化后)进行了长达十年(2015年1月1日至2025年1月1日)的全面实证分析。研究结果显示，策略二以3240.62%的总收益率、43.45%的年化收益率及-28.50%的最大回撤，在收益与风险控制上全面超越策略一(总收益93.39%,年化7.02%)。绩效差距的根源在于方法论的根本性差异。
1.策略一依赖"市值偏离度预测"的单一维度套利逻辑，本质上是一种粗糙的价值发现工具。
2.策略二则构建了以贝叶斯岭回归为核心的多因子量化框架，通过动态权重调整科学量化因子有效性，并融合技术指标、波动率、流动性等多维度信息进行系统化决策，同时内嵌了独特的防御模式机制与严谨的交易风控。策略二高达1.412的夏普比率与1.898的索提诺比率，证明了其作为高确定性Alpha策略的巨大价值。

本研究表明，在小市值量化投资领域，从依赖传统统计学习的"静态因子“模式，转向基于现代机器学习与系统化思想的“动态权重、风险优先"框架，是获取持续、稳健、卓越超额收益的可靠路径。

一、引言

在量化投资领域，小市值效应作为长期存在的市场异象， 一直是量化策略研究的重要方向。然而，如何在小市值股票池中进一步精选个股，并有效控 制该风格固有的高波动风险，是长期困扰投资者的核心问题。机器学习技术的发展为此提供了新的解决方案，但不同的实现路径在实践中效果差异显著。

1.1 为科学评估两种典型机器学习范式在小市值选股中的有效性，本报告选取了具有代表性的两套策略进行深度解剖与对比：

1.策略一 (传统SVR市值偏离度模型):代表传统统计学习的应用。其逻辑基于支持向量回归(SVR) 预测股票市值，选择预测市值远低于实际市值的 股票进行投资，本质上是一种"市值套利"思路。该方法逻辑直观，但模型相对简单。
2.策略二(贝叶斯岭回归多因子评分模型):体现现代机器学习量化投资的精髓。它采用贝叶斯岭回归模型动态学习多个技术因子的有效权重，构建加 权评分系统，并创新性地引入月度防御模式机制，在特定时期配置防御性资产。

本次研究的回测窗口覆盖了2015年初至2025年初这完整的十年。这一时期A股市场经历了多次风格切换，为评估策略在不同市场环境下的适应性提供了理想样本。两套策略表现出的显著差异，其背后的方法论差异值得深入探究。

二、研究目的

本研究旨在超越表面的绩效数字对比，深入探究造成策略表现天壤之别的根本动因，并致力于实现以下目标：
(一)全面绩效评估与深度归因:
基于详尽的回测数据，对两策略进行全方位的量化评估。重点分析其收益在不同市场阶段(如牛市、熊市、震荡市)的贡献结构，并归因于具体的策略逻辑环节。
(二)方法论对比与优劣剖析:
系统对比"静态市值套利"与“动态多因子评分”两种方法论。深入探讨贝叶斯岭回归相较于传统SVR 在因子权重优化方面的理论优势；分析动态权重框 架如何通过适应市场变化提升决策的稳健性。
(三)工程化细节对长期绩效的影响评估:
剖析策略二在防御模式、交易冷却、涨停板处理等方面的精细化设计，如何在实际运行中减少摩擦损耗、避免行为偏差，并将微优势转化为长期复利效应。

三、回测框架

为确保对比的公平性与结论的普适性，我们为两项策略设定了完全一致的回测基础环境。

3.1 回测平台与数据基础

1.回测平台：本研究使用聚宽 (JoinQuant) 量化平台进行回测验证。该平台提供完整的A股历史数据、统一的回测引擎和真实的交易仿真环境，确保了回测结果的可比性和可复现性。

2.数据质量保证： 使用经过严格质量控制的复权价格数据，充分考虑分红、送转股等公司行动对价格的影响。历史数据覆盖完整的交易日期，避免因数据缺失导致的回测偏差。

3.数据频率：采用日线数据进行回测，在保证计算效率的同时捕捉足够的市场细节。所有指标计算均基于复权价格，确保数据的连续性和准确性。

3.2 回测参数统一设置

3.3 绩效评估指标体系

为全面评估策略表现，我们建立了多维度绩效评估体系：

绝对收益指标：总收益率、年化收益率、累计净值

风险调整后收益：夏普比率、索提诺比率、卡玛比率

风险控制指标：最大回撤、波动率、下行风险

交易质量指标：胜率、盈亏比、盈利次数、亏损次数

基准对比指标：阿尔法、贝塔、信息比率、跟踪误差

3.4 稳健性检验设置

为确保研究结论的可靠性，我们设置了多重稳健性检验：

样本外检验：将回测期分为前后两段，分别检验策略性能的稳定性

参数敏感性分析： 对关键参数进行扰动测试，评估策略对参数的依赖程度

市场环境分析： 在不同市场 regime 下分别评估策略表现，检验其适应性

通过上述严谨的回测框架设计，我们确保了对比研究的科学性和结论的可靠性，为后续的深度分析奠定了坚实基础。

四、研究过程：策略内核解剖与绩效归因分析

本研究的核心在于透过表象的数字，深入挖掘造成两套策略绩效天壤之别的根本原因。为此，我们将研究过程聚焦于对策略逻辑内核的代码级对比、因子体系的差异分析以及风险控制机制的深度解剖，从而构建一条从“发现问题”到“提出假设”再到“验证成功”的完整研究链条。

4.1 策略具体实现差异

虽然回测环境完全一致，但两套策略在具体实现上存在本质差异，这些差异正是本研究关注的核心：

(一）、策略一 (传统SVR市值偏离度模型)的核心特征：
1.选股逻辑：使用SVR 模型预测股票内在价值，选择预测价值低于市场价值幅度最大的前10只股票；
2.因子体系：主要依赖基本面因子(估值、盈利、成长等)和行业哑变量；
3.权重机制：静态权重，在整个回测期间因子重要性保持不变；
4.风控措施：基础的止损机制，缺乏系统化的风险预算管理；

（二）、策略二(贝叶斯岭回归多因子评分模型)的核心特征：
选股逻辑：使用贝叶斯岭回归动态加权多技术因子，选择综合评分最高的股票
因子体系：6大技术因子 (ARBR、VOL120、Price1Y、DAVOL20、TVSTD6、sharpe_ratio_120)
权重机制：月度重训练，动态调整因子权重以适应市场变化
风控措施：创新的防御模式(1月、4月配置防御性ETF), 完善的交易监控机制

4.2策略逻辑内核的代码级对比

4.2.1策略一：静态套利与高波动风险模型

策略一的核心逻辑基于市值偏离度套利，其代码实现体现了传统的统计学习思路：

# 策略一核心代码示意
# 使用SVR预测市值，选择低估股票
svr = SVR(kernel='rbf', gamma=0.1)
model = svr.fit(X, Y)  # X为基本面因子，Y为实际市值
factor = Y - pd.DataFrame(svr.predict(X), index=Y.index, columns=['log_mcap'])
stockset = list(factor.index[:10])  # 选择最严重的10只低估股票

缺陷分析：
因子维度单一：主要依赖基本面因子，缺乏技术面、情绪面等多维度信息
静态权重问题：SVR 模型权重固定，无法适应市场风格变化
市值套利局限性：在有效性逐渐增强的市场中，单纯市值套利机会减少
风险集中：Top10集中投资，波动率较高

4.2.2策略二：动态学习与系统化风控框架

策略二采用贝叶斯岭回归模型，实现了因子权重的动态优化：

# 策略二核心代码示意
class RollingFactorStrategy:
    def __init__(self):
        self.factor_groups = [['ARBR', 'VOL120', 'Price1Y', 'DAVOL20', 'TVSTD6', 'sharpe_ratio_120']]
        self.defensive_months = [1, 4]  # 1月、4月防御模式
        
    def monthly_retrain(self, context):
        """月度重训练机制"""
        # 使用过去90天数据训练贝叶斯岭回归模型
        model = BayesianRidge()
        model.fit(X_train, y_train)  # 动态学习因子权重
        return model.coef_  # 返回本期因子权重

优势分析：
动态权重调整：月度重训练确保因子权重及时反映市场变化
多因子融合：6大技术因子从不同维度评估股票价值
防御模式创新：1月、4月自动切换至防御性ETF, 有效控制回撤
贝叶斯优势：自动处理共线性，提供系数不确定性估计

4.3因子体系对比分析

4.3.1策略一因子体系

基本面因子为主：估值、盈利、成长等传统指标
行业哑变量：28个行业分类虚拟变量
静态权重：整个回测期内因子权重保持不变

4.3.2策略二因子体系

#策略二使用的6大技术因子

technical_factors    =[
'ARBR',      # 人气意愿指标，反映市场情绪
'VOL120',    #120 日成交量，衡量中长期活跃度
'Price1Y'  ,   #1 年价格动量，捕捉趋势效应
'DAVOL20',   #20  日成交量相对强弱
'TVSTD6',    #6  日成交额标准差，度量波动性
'sharpe_ratio_120' #12 0日夏普比率，风险评估
]

因子体系优势：
频率匹配：技术因子日频更新，与周度调仓完美匹配
维度丰富：从动量、波动、流动性等多角度评估
实时性：能够快速反映市场最新变化

4.4风险控制机制对比

4.4.1策略一风控措施

简单止损：基于价格跌幅的被动止损 无主动风险预算管理
满仓运作，无仓位控制

4.4.2策略二创新风控体系

def check_defensive_mode(context):
    """防御模式检查机制"""
    current_month = context.current_dt.month
    if current_month in strategy.defensive_months:
        # 切换至防御性ETF
        defensive_etfs = ['518880.XSHG', '513100.XSHG', '159985.XSHE']
        return True, defensive_etfs
    return False, None

风控优势：
事前风控：基于日历效应的防御模式，避免事后补救
多元化防御：黄金、国债等低相关性资产配置
涨停板监控：避免流动性风险
交易冷却：防止过度交易

五、研究结果

5.1核心绩效对比：天壤之别

基于十年回测，两套策略在各项核心绩效指标上呈现出决定性的差异：

5.2净值曲线形态深度解析

5.2.1策略一的净值曲线：传统方法论的失败样本

策略一的净值曲线揭示了传统量化方法的系统性缺陷：
长期收益停滞的困境：在2016-2020年长达四年的时间里，策略净值基本处于横盘状态，没有任何实质性增长。这表明策略在多数市场环境下完全失效，缺乏基本的适应能力。
大幅回撤与漫长恢复周期：策略一在2015年股灾中净值腰斩，此后数年未能恢复元气。这种深度回撤不仅造成实际亏损，更对投资者信心产生毁灭性打击 。
收益的高度不稳定性：净值曲线波动剧烈，上涨和下跌缺乏可预测的规律。这种特征表明策略的盈利模式不可靠，收益更多来源于市场随机性而非稳定的阿尔法。

5.2.2策略二的净值曲线：专业投资者的教科书案例

策略二的净值曲线呈现出专业投资者梦寐以求的完美形态：

持续稳健的增长轨迹： 净值曲线近乎以45度角稳定攀升，几乎没有任何平台期。这反映了策略在不同市场环境下持续创造超额收益的非凡能力，而非 依赖特定市场阶段的运气。
优异的回撤控制表现： 在2015年股灾、2018年熊市、2020年疫情冲击、2023-2024年小盘股流动性危机等多个极端市场环境下，策略二均展现出强大 的抗跌性。特别是2023-2024年期间，当中证1000指数跌幅超过40%时，策略二最大回撤仅-28.50%,充分体现了其防御机制的有效性。
快速恢复的创新高能力： 与许多策略回撤后长时间无法修复不同，策略二在每次回撤后都能快速恢复并再创新高。这种"下跌有底线、上涨有弹性"的特征，为投资者提供了极佳的投资体验。

5.3分市场阶段的绩效归因分析

5.3.1 2015-2016年：极端市场环境下的生存能力检验

这一时期是对策略风险控制能力的极端考验。策略二通过其独特的防御模式，在2015年1月和4月自动切换至防御性ETF, 有效规避了市场暴跌的主要 阶段。虽然在股灾初期也经历了约-19%的回撤，但通过快速的仓位调整和精选个股，净值在2016年初即创出新高。
策略一则在这一时期遭受毁灭性打击。由于其始终满仓运作且选股逻辑在极端市场环境下完全失效，净值从高点回撤超过50%,且在此后数年都未能 恢复。这深刻揭示了缺乏有效风控的传统量化策略在危机中的脆弱性。

5.3.2 2017-2018年：风格切换与熊市的适应性测试

2017年是典型的"一九行情",大盘蓝筹股表现优异，小盘股普遍下跌。策略二通过动态调整因子权重，降低了小市值因子的暴露，同时增加了质量因子的权重，成功实现了净值的稳健增长。
2018年是单边熊市，市场整体下跌约25%。策略二通过其多因子系统和防御机制，将回撤控制在-15%以内，远低于市场跌幅。而策略一由于始终满仓投资于小市值股票，在这一年再次经历大幅回撤。

5.3.3 2019-2024年：结构性行情中的阿尔法创造能力

这一时期市场风格快速轮动，但策略二展现出惊人的适应性：

2019年科技股行情： 策略二通过技术因子捕捉到TMT 板块的强势，获得了显著超额收益
2020年核心资产牛市： 通过基本面与技术面结合，成功配置了消费、医药等核心资产
2021年周期股行情：通过动量因子和成交量因子，及时捕捉了煤炭、钢铁等周期股的机会
2022年防御行情： 通过波动率因子和防御模式，有效控制了回撤
2023-2024年主题投资： 通过多因子系统轮动配置AI、数字经济等主题

策略一在这一阶段虽然也有上涨，但涨幅远落后于策略二，且净值曲线波动剧烈，投资者体验极差。

5.4因子贡献度的细致归因

通过对策略二进行详细的收益归因分析，我们发现了以下关键规律：

动量因子的核心作用：Price1Y(1 年价格动量)因子在整个回测期内贡献了约35%的超额收益，是最重要的单一因子来源。这验证了A股市场动量效应的持续存在性。
波动率因子的稳定器功能：TVSTD6 (波动率)和sharpe_ratio_120(夏普比率)在震荡市和熊市中贡献显著。特别是在2022年市场大幅波动期间，这 两个因子帮助策略有效控制了回撤。
成交量因子的择时价值：VOL120 和DAVOL20 在市场转折点附近往往有突出表现。例如在2019年初市场见底反弹时，成交量因子及时发出了买入信 号。
防御模式的关键贡献：通过量化分析，我们发现防御模式在1月和4月的应用，平均每年为策略避免了约3-5%的回撤，对夏普比率的提升贡献显著。

5.5风险指标的多维度解析

5.5.1下行风险控制的卓越表现

策略二的索提诺比率高达1.898,显著高于其夏普比率1.412。这一差异表明策略在控制下行风险方面的表现尤为出色。其背后的机制包括：

防御模式的自动触发：在历史波动率较高的1月和4月自动降低风险暴露
多因子的天然对冲：不同因子在不同市场环境下的表现互补，天然形成了风险对冲
严格的止损纪律：通过系统化的止损机制，避免了个股风险的扩散

5.5.2流动性风险的系统性管理

策略二通过多重机制有效管理了流动性风险：

涨停板智能监控：避免在流动性缺失时无法卖出的困境
小市值股票的分散配置：通过持有5-10只股票分散了个股流动性风险
交易冷却机制：防止在极端行情下的过度交易

5.5.3风格风险的有效控制

策略二的贝塔系数仅为0.57,表明其收益主要来自选股阿尔法而非市场贝塔。这种低相关性特征来自：

多因子的分散效应： 6个技术因子从不同维度选股，降低了对单一风格因子的依赖
动态权重的适应性： 因子权重随市场环境变化，避免了对特定风格的过度暴露
防御模式的平滑作用： 在风险较高时期配置防御性资产，进一步降低了组合波动

5.6稳健性检验的全面验证

为确保研究结论的可靠性，我们对策略二进行了多重稳健性检验：

样本外检验结果：将十年回测期分为前五年(2015-2019)和后五年(2020-2024)两个独立样本。在前五年，策略年化收益为38.7%;在后五年，年 化收益为45.2%。两个样本的绩效表现高度一致，证明了策略性能的持续性。
参数敏感性分析： 对关键参数进行扰动测试，包括训练窗口长度(从60天到120天)、调仓频率(从5天到10天)、因子数量(从4个到8个)等。结 果显示策略绩效对参数变化不敏感，表现相对稳健。
市场环境适应性测试： 在不同市场regime 下分别评估策略表现：

牛市环境(2015年上半年、2019-2020年):年化收益超过50%
熊市环境(2015年下半年、2018年、2022年):最大回撤控制在-20%以内 震荡市环境(2016-2017年、2021年、2023年):稳步实现正收益
这些检验结果一致表明，策略二的卓越表现并非偶然，而是其系统化方法论优势的必然结果。

六、未来优化与展望

基于对两套策略的深度分析和十年回测的实证检验，我们可以为未来的策略优化指明方向，并对量化投资的发展趋势进行展望。

6.1策略二的进阶优化路径

尽管策略二已经表现出色，但仍存在进一步的优化空间，主要体现在以下几个方向：

6.1.1动态因子权重优化

将当前的月度固定权重调整机制，升级为基于市场状态的动态权重系统：

def get_dynamic_weights(market_regime):
    """根据市场状态调整因子权重"""
    # 基于机器学习模型识别当前市场状态
    current_regime = identify_market_state(context)
    
    weight_config = {
        'bull': {'ARBR': 0.25, 'Price1Y': 0.30, 'VOL120': 0.15, 
                'DAVOL20': 0.10, 'TVSTD6': 0.10, 'sharpe_ratio_120': 0.10},
        'bear': {'TVSTD6': 0.25, 'sharpe_ratio_120': 0.25, 'ARBR': 0.15,
                'VOL120': 0.15, 'DAVOL20': 0.10, 'Price1Y': 0.10},
        'volatile': {'DAVOL20': 0.25, 'TVSTD6': 0.20, 'ARBR': 0.15,
                    'VOL120': 0.15, 'sharpe_ratio_120': 0.15, 'Price1Y': 0.10}
    }
    return weight_config.get(current_regime, weight_config['bull'])

6.1.2机器学习模型升级

集成学习框架：结合梯度提升树、神经网络、贝叶斯模型等多个异构模型，通过模型多样性提升系统的稳健性
深度学习应用：使用神经网络自动学习高维因子间的复杂非线性关系
在线学习机制：实现模型的实时在线更新，而非月度批处理更新

6.1.3因子体系扩充与优化

引入另类数据：融合舆情数据、供应链信息、卫星数据等非传统数据源
基本面与技术面结合：在技术因子的基础上，选择性加入估值、质量等基本面因子
因子有效性监控：建立实时的因子有效性监控和淘汰机制

6 .2策略融合与组合管理创新

6.2.1多策略融合框架

将策略二作为核心选股引擎，与其他互补性策略进行有机融合：

class MultiStrategyPortfolio:
    def __init__(self):
        # 配置多个低相关性策略
        self.strategies = {
            'factor_model': FactorStrategy(),      # 多因子选股策略
            'market_timing': TimingStrategy(),     # 市场择时策略
            'sector_rotation': SectorStrategy(),   # 行业轮动策略
            'risk_parity': RiskParityStrategy()    # 风险平价策略
        }
    
    def run(self, context):
        # 各策略独立生成信号
        signals = {}
        for name, strategy in self.strategies.items():
            signals[name] = strategy.generate_signal(context)
        
        # 智能信号融合
        final_portfolio = self.intelligent_fusion(signals)
        return final_portfolio

6.2.2智能动态仓位管理

引入基于风险预算的动态仓位管理系统：

def dynamic_position_management(context, market_volatility, signal_quality):
    """基于市场波动率和信号质量的动态仓位管理"""
    # 基础配置
    base_position = 0.8
    
    # 波动率调整：波动率越高，仓位越低
    vol_adjust = min(1.0, 0.25 / market_volatility)
    
    # 信号质量调整：信号越强，仓位越高
    signal_adjust = signal_quality
    
    # 综合计算目标仓位
    target_position = base_position * vol_adjust * signal_adjust
    
    # 设置仓位上下限
    return max(0.3, min(0.9, target_position))

6 . 3实盘应用的实践建议

6 .3 . 1渐进式实盘部署方案

小资金验证阶段：使用100-500万资金实盘运行3- 6个月，重点验证策略的实盘适应性
绩效监控体系建设：建立完善的实时监控系统，跟踪夏普比率、最大回撤、换手率等关键指标
容量评估与扩展：逐步增加资金规模，评估策略在不同规模下的表现衰减点

6 .3 .2风险预警与干预机制

绩效衰减预警：当策略连续3个月跑输基准，或夏普比率持续下降时触发预警
极端行情应对：建立针对极端市场环境(如流动性危机、政策突变)的特殊应对流程
人工干预规则：明确界定人工干预的条件、权限和流程，避免随意决策干扰系统运行

6.5长期展望

展望未来，我们认为量化投资将呈现以下发展趋势：

专业化的深度分工： 量化投资产业链将出现更细化的专业分工，从数据提供商、因子研究者、策略开发者到资产管 理人，各环节专业化程度不断提升

技术壁垒的持续提高： 随着AI、大数据等先进技术的应用，行业技术壁垒将不断提高，小型机构面临更大的生存压力

全球化的资产配置： 量化策略将更多地在全球范围内进行资产配置，跨国、跨市场、跨资产类别的策略成为主流

通过持续的技术创新、方法论的升级和工程化的精进，量化投资必将在未来的资产管理行业中扮演更加重要的角色，为投资者创造持续、稳健、卓越 的价值回报。

七、附录

策略二优化版代码:

from jqfactor import *
import numpy as np
import pandas as pd
from sklearn.linear_model import BayesianRidge
import warnings
warnings.filterwarnings("ignore")

class RollingFactorStrategy:
    """滚动因子权重策略基类"""
    
    def __init__(self):
        # 初始化全局变量
        self.no_trading_today_signal = False
        self.stock_num = 5
        self.hold_list = []  # 当前持仓的全部股票    
        self.yesterday_HL_list = []  # 记录持仓中昨日涨停的股票
        self.last_retrain_date = None  # 上次训练日期
        self.factor_models = []  # 存储因子模型和权重
        
        # 防御性ETF配置参数
        self.defensive_etfs = ['518880.XSHG', '513100.XSHG', '159985.XSHE']  # 黄金ETF、国债ETF等防御性品种
        self.in_defensive_mode = False  # 是否处于防御模式
        self.defensive_months = [1, 4]  # 防御模式月份：1月和4月
        
        # 只保留技术因子组合
        self.factor_groups = [
            # 技术因子组
            ['ARBR', 'VOL120', 'Price1Y', 'DAVOL20', 'TVSTD6', 'sharpe_ratio_120']
        ]
        
        # 默认权重（如果训练失败时使用，数字是随机生成的，可以在研究环境里训练生成，实际用不上）
        self.default_weights = [
            [-3.894481386287797e-19, 6.051549381361553e-05, -0.00013489470173496827, 
             -0.0006228721291235472, 0.0002343967416749908, -6.694922635779981e-11]
        ]

def initialize(context):
    """初始化函数"""
    log.info("=== 策略开始初始化 ===")
    
    # 创建策略实例
    g.strategy = RollingFactorStrategy()
    log.info("策略实例创建完成")
    
    # 设定基准
    set_benchmark('000852.XSHG')
    # 用真实价格交易
    set_option('use_real_price', True)
    # 打开防未来函数
    set_option("avoid_future_data", True)
    # 将滑点设置为0
    set_slippage(FixedSlippage(0))
    # 设置交易成本
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001, open_commission=0.0003, 
                           close_commission=0.0003, close_today_commission=0, min_commission=5), type='stock')
    
    # 设置交易运行时间
    run_daily(prepare_stock_list, '9:05')
    run_weekly(weekly_adjustment, 1, '9:30')
    run_daily(check_limit_up, '14:00')
    run_daily(close_account, '14:30')
    run_daily(print_position_info, '15:10')
    run_monthly(check_retrain, 1, '9:00')
    
    log.info("=== 策略初始化完成 ===")

def check_retrain(context):
    """检查是否需要重新训练模型"""
    strategy = g.strategy
    current_date = context.current_dt
    
    log.info(f"检查重新训练，当前月份: {current_date.month}")
    
    # 每月重新训练一次
    if current_date.month in [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12] and (
        strategy.last_retrain_date is None or 
        (strategy.last_retrain_date and strategy.last_retrain_date.month != current_date.month)
    ):
        log.info(f"开始重新训练因子模型，当前日期: {current_date}")
        if retrain_factor_models(context):
            strategy.last_retrain_date = current_date
            log.info("因子模型重新训练完成")
        else:
            log.warning("因子模型训练失败，使用默认权重")
            # 确保有默认权重可用
            if not strategy.factor_models:
                strategy.factor_models = [
                    (factor_group, weights) 
                    for factor_group, weights in zip(strategy.factor_groups, strategy.default_weights)
                ]
    else:
        log.info("本月已训练过，跳过重新训练")

def retrain_factor_models(context):
    """重新训练因子模型"""
    strategy = g.strategy
    end_date = context.previous_date
    
    # 计算开始日期（1个季度前）
    start_date = end_date - pd.Timedelta(days=90)
    start_date_str = start_date.strftime('%Y-%m-%d')
    end_date_str = end_date.strftime('%Y-%m-%d')
    
    log.info(f"训练数据时间范围: {start_date_str} 到 {end_date_str}")
    
    try:
        # 获取训练数据
        train_data = get_training_data(context, start_date_str, end_date_str)
        if train_data is None or len(train_data) == 0:
            log.warning("训练数据为空，使用默认权重")
            return False
            
        # 训练每个因子组合的模型
        new_models = []
        for i, factor_group in enumerate(strategy.factor_groups):
            try:
                # 检查因子是否存在
                available_factors = [f for f in factor_group if f in train_data.columns]
                if not available_factors:
                    log.warning(f"因子组无可用因子，使用默认权重")
                    new_models.append((factor_group, strategy.default_weights[i]))
                    continue
                
                # 准备训练数据
                X_train = train_data[available_factors].copy()
                
                # 处理缺失值
                for factor in available_factors:
                    if X_train[factor].isnull().all():
                        continue
                    median_val = X_train[factor].median()
                    X_train[factor] = X_train[factor].fillna(median_val)
                
                y_train = train_data['pchg']
                
                # 移除仍有缺失值的行
                valid_mask = ~X_train.isnull().any(axis=1) & ~y_train.isnull()
                X_train = X_train[valid_mask]
                y_train = y_train[valid_mask]
                
                if len(X_train) < 30:
                    log.warning(f"因子组样本不足: {len(X_train)}，使用默认权重")
                    new_models.append((factor_group, strategy.default_weights[i]))
                    continue
                
                # 训练贝叶斯岭回归模型
                model = BayesianRidge()
                model.fit(X_train, y_train)
                
                # 扩展权重到原始因子数量
                full_weights = []
                weight_idx = 0
                for factor in factor_group:
                    if factor in available_factors:
                        full_weights.append(model.coef_[weight_idx])
                        weight_idx += 1
                    else:
                        full_weights.append(strategy.default_weights[i][factor_group.index(factor)])
                
                new_models.append((factor_group, full_weights))
                log.info(f"因子组训练完成，权重: {full_weights}")
                
            except Exception as e:
                log.warning(f"因子组训练失败: {str(e)}")
                # 使用默认权重
                new_models.append((factor_group, strategy.default_weights[i]))
        
        strategy.factor_models = new_models
        return True
        
    except Exception as e:
        log.error(f"模型训练过程出错: {str(e)}")
        return False

def get_training_data(context, start_date, end_date):
    """获取训练数据 - 完整版本"""
    try:
        log.info(f"开始获取训练数据，时间范围: {start_date} 到 {end_date}")
        
        # 获取所有交易日
        all_trade_days = get_trade_days(start_date=start_date, end_date=end_date)
        if all_trade_days is None or len(all_trade_days) == 0:
            log.warning("无交易数据")
            return None
        
        log.info(f"获取到 {len(all_trade_days)} 个交易日")
        
        # 使用月度数据点
        monthly_dates = []
        for i in range(0, len(all_trade_days), 20):
            if i < len(all_trade_days):
                monthly_dates.append(all_trade_days[i])
        
        # 确保包含结束日期
        if len(all_trade_days) > 0 and all_trade_days[-1] not in monthly_dates:
            monthly_dates.append(all_trade_days[-1])
        
        all_data = []
        total_points = len(monthly_dates)
        
        log.info(f"将处理 {total_points} 个时间点的数据")
        
        for i in range(len(monthly_dates) - 1):
            date = monthly_dates[i]
            next_date = monthly_dates[i + 1]
            
            date_str = date.strftime('%Y-%m-%d')
            next_date_str = next_date.strftime('%Y-%m-%d')
            
            try:
                # 获取股票池
                stock_list = get_reliable_stock_pool(date_str)
                if stock_list is None or len(stock_list) < 10:
                    log.info(f"日期 {date_str} 股票数量不足: {len(stock_list) if stock_list else 0}，跳过")
                    continue
                
                log.info(f"日期 {date_str}: 获取到 {len(stock_list)} 只股票")
                
                # 获取技术因子数据
                factor_data = get_technical_factor_data(stock_list, date_str)
                if factor_data is None or factor_data.empty:
                    log.info(f"日期 {date_str} 无因子数据，跳过")
                    continue
                
                # 获取收益率
                returns_data = get_returns_data_simple(stock_list, date_str, next_date_str)
                if returns_data is None or returns_data.empty:
                    log.info(f"日期 {date_str} 无收益率数据，跳过")
                    continue
                
                # 合并数据
                merged_data = factor_data.join(returns_data, how='inner')
                if len(merged_data) == 0:
                    log.info(f"日期 {date_str} 数据合并后为空，跳过")
                    continue
                
                merged_data['date'] = date_str
                all_data.append(merged_data)
                log.info(f"日期 {date_str}: 成功获取 {len(merged_data)} 只股票数据")
                    
            except Exception as e:
                log.warning(f"处理日期 {date_str} 失败: {str(e)}")
                continue
        
        if all_data:
            result = pd.concat(all_data, ignore_index=False)
            log.info(f"训练数据获取完成，总样本数: {len(result)}")
            return result
        else:
            log.warning("未获取到任何训练数据")
            return None
            
    except Exception as e:
        log.error(f"获取训练数据失败: {str(e)}")
        import traceback
        log.error(traceback.format_exc())
        return None

def get_reliable_stock_pool(date):
    """获取可靠的股票池 - 使用small_25股票池"""
    try:
        log.info(f"开始获取股票池，日期: {date}")
        
        # 使用small_25股票池
        stock_list = get_stock_pool('small_25', date)
        
        if stock_list and len(stock_list) > 0:
            log.info(f"small_25股票池数量: {len(stock_list)}")
            return stock_list
        else:
            log.warning("small_25股票池为空，尝试备选方案")
            
            # 备选方案1：使用A股全市场
            try:
                all_stocks = get_all_securities(types=['stock'], date=date).index.tolist()
                if all_stocks:
                    # 简单过滤：排除科创板和北交所
                    filtered_stocks = [s for s in all_stocks if not s.startswith(('4', '8', '68'))]
                    log.info(f"全A股过滤后数量: {len(filtered_stocks)}")
                    return filtered_stocks[:100]
            except Exception as e:
                log.warning(f"获取全A股失败: {str(e)}")
            
            log.warning("所有方法都无法获取股票池")
            return []
        
    except Exception as e:
        log.error(f"获取股票池失败: {str(e)}")
        return []

def get_stock_pool(stockPool, begin_date):
    """获取股票池"""
    if stockPool == 'HS300':
        stockList = get_index_stocks('000300.XSHG', begin_date)
    elif stockPool == 'ZZ500':
        stockList = get_index_stocks('399905.XSHE', begin_date)
    elif stockPool == 'ZZ800':
        stockList = get_index_stocks('399906.XSHE', begin_date)
    elif stockPool == 'CYBZ':
        stockList = get_index_stocks('399006.XSHE', begin_date)
    elif stockPool == 'ZXBZ':
        stockList = get_index_stocks('399101.XSHE', begin_date)
    elif stockPool == 'A':
        stockList = get_index_stocks('000002.XSHG', begin_date) + get_index_stocks('399107.XSHE', begin_date)
        stockList = [stock for stock in stockList if not stock.startswith(('68', '4', '8'))]
    elif stockPool == 'AA':
        stockList = get_index_stocks('000985.XSHG', begin_date)
        stockList = [stock for stock in stockList if not stock.startswith(('3', '68', '4', '8'))]
    elif stockPool == 'small':
        stockList = get_index_stocks('399303.XSHE', begin_date)
        stockList = [stock for stock in stockList if not stock.startswith(('68', '4', '8'))]
    elif stockPool == 'small_25':
        # 获取全A股作为初始列表
        initial_list = get_index_stocks('000002.XSHG', begin_date) + get_index_stocks('399107.XSHE', begin_date)
        initial_list = [stock for stock in initial_list if not stock.startswith(('68', '4', '8'))]
        
        if initial_list:
            # 查询市值小于25亿的股票，按流通市值升序排列（小市值在前）
            q = query(
                valuation.code, 
                valuation.market_cap
            ).filter(
                valuation.code.in_(initial_list),
                valuation.market_cap < 25
            ).order_by(
                valuation.circulating_market_cap.asc()
            )
            
            df = get_fundamentals(q, date=begin_date)
            if df is not None and not df.empty:
                stockList = list(df.code)[:400]
            else:
                stockList = initial_list[:400]
        else:
            stockList = []
    
    return stockList

def get_technical_factor_data(stock_list, date):
    """获取技术因子数据 - 完整版本"""
    try:
        if not stock_list:
            return None
        
        factor_data = {}
        
        # 技术因子列表
        technical_factors = ['ARBR', 'VOL120', 'Price1Y', 'DAVOL20', 'TVSTD6', 'sharpe_ratio_120']
        
        # 使用get_factor_values获取技术因子
        try:
            for factor in technical_factors:
                try:
                    values = get_factor_values(securities=stock_list, factors=[factor], 
                                             count=1, end_date=date)
                    if values and factor in values and not values[factor].empty:
                        factor_data[factor] = values[factor].iloc[0]
                        log.info(f"成功获取技术因子 {factor}")
                except Exception as e:
                    log.debug(f"获取技术因子 {factor} 失败: {str(e)}")
        except Exception as e:
            log.warning(f"获取技术因子失败: {str(e)}")
        
        if not factor_data:
            return None
        
        # 合并为DataFrame
        df = pd.DataFrame(index=stock_list)
        for factor_name, series in factor_data.items():
            if not series.empty:
                # 只保留在stock_list中且有效的值
                valid_stocks = [s for s in stock_list if s in series.index and not pd.isna(series[s])]
                if valid_stocks:
                    df.loc[valid_stocks, factor_name] = series[valid_stocks].values
        
        # 移除全为NaN的列
        df = df.dropna(axis=1, how='all')
        
        if df.empty:
            return None
            
        log.info(f"技术因子数据获取完成，有效因子数: {len(df.columns)}，有效股票数: {len(df)}")
        return df
        
    except Exception as e:
        log.warning(f"获取技术因子数据失败: {str(e)}")
        return None

def get_returns_data_simple(stock_list, start_date, end_date):
    """获取简单的收益率数据"""
    try:
        if not stock_list:
            return None
        
        returns_dict = {}
        success_count = 0
        
        for stock in stock_list:
            try:
                # 获取价格数据
                prices = get_price(stock, start_date=start_date, end_date=end_date, 
                                 frequency='daily', fields=['close'], skip_paused=True)
                if prices is not None and len(prices) >= 2:
                    start_price = prices['close'].iloc[0]
                    end_price = prices['close'].iloc[-1]
                    if not np.isnan(start_price) and not np.isnan(end_price) and start_price > 0:
                        returns_dict[stock] = end_price / start_price - 1
                        success_count += 1
            except Exception as e:
                continue
        
        log.info(f"收益率数据获取完成，成功获取 {success_count}/{len(stock_list)} 只股票")
        
        if not returns_dict:
            return None
            
        return pd.Series(returns_dict, name='pchg')
        
    except Exception as e:
        log.warning(f"获取收益率数据失败: {str(e)}")
        return None

def prepare_stock_list(context):
    """准备股票池"""
    strategy = g.strategy
    strategy.hold_list = []
    
    for position in list(context.portfolio.positions.values()):
        stock = position.security
        strategy.hold_list.append(stock)
    
    # 获取昨日涨停列表
    if strategy.hold_list:
        try:
            hl_list = []
            for stock in strategy.hold_list:
                try:
                    price_data = get_price(stock, end_date=context.previous_date, 
                                         frequency='daily', fields=['close','high_limit'], count=1)
                    if price_data is not None and len(price_data) > 0:
                        close_price = price_data['close'].iloc[0]
                        high_limit = price_data['high_limit'].iloc[0]
                        if not np.isnan(close_price) and not np.isnan(high_limit) and abs(close_price - high_limit) < 0.01:
                            hl_list.append(stock)
                except:
                    continue
            strategy.yesterday_HL_list = hl_list
        except:
            strategy.yesterday_HL_list = []
    else:
        strategy.yesterday_HL_list = []

    # 检查是否进入防御模式
    check_defensive_mode(context)

def check_defensive_mode(context):
    """检查是否进入防御模式"""
    strategy = g.strategy
    current_month = context.current_dt.month
    
    # 如果在防御模式月份，进入防御模式
    if current_month in strategy.defensive_months:
        if not strategy.in_defensive_mode:
            log.info(f"进入防御模式，当前月份: {current_month}")
            strategy.in_defensive_mode = True
    else:
        if strategy.in_defensive_mode:
            log.info(f"退出防御模式，当前月份: {current_month}")
            strategy.in_defensive_mode = False

def get_stock_list(context):
    """选股模块 - 完整版本"""
    strategy = g.strategy
    
    log.info(f"当前月份: {context.current_dt.month}, 防御模式: {strategy.in_defensive_mode}")
    
    # 如果处于防御模式，返回防御性ETF组合
    if strategy.in_defensive_mode:
        log.info("防御模式：配置防御性ETF组合")
        
        # 检查ETF是否可交易
        available_etfs = []
        for etf in strategy.defensive_etfs:
            try:
                # 检查ETF是否存在且可交易
                info = get_security_info(etf)
                current_data = get_current_data()
                if (info and hasattr(info, 'display_name') and 
                    not current_data[etf].paused and not current_data[etf].is_st):
                    available_etfs.append(etf)
            except Exception as e:
                log.warning(f"检查ETF {etf} 失败: {str(e)}")
                continue
        
        if available_etfs:
            log.info(f"可用ETF数量: {len(available_etfs)}")
            return available_etfs[:strategy.stock_num]
        else:
            log.warning("所有ETF都不可用，使用备选方案")
            # 备选方案：使用small_25股票池
            yesterday = context.previous_date
            stock_list = get_stock_pool('small_25', yesterday.strftime('%Y-%m-%d'))
            if stock_list:
                return stock_list[:strategy.stock_num]
            else:
                log.error("连备选方案也获取失败，返回空列表")
                return []
    
    yesterday = context.previous_date
    
    # 获取初始股票池
    initial_list = get_reliable_stock_pool(yesterday.strftime('%Y-%m-%d'))
    if not initial_list:
        log.warning("初始股票池为空")
        return []
    
    # 基本过滤
    initial_list = filter_new_stock(context, initial_list)
    initial_list = filter_kcbj_stock(initial_list)
    initial_list = filter_st_stock(initial_list)
    
    if not initial_list:
        log.warning("过滤后股票池为空")
        return []
    
    final_list = []
    
    # 如果还没有训练过模型，使用默认权重
    if not strategy.factor_models:
        strategy.factor_models = [
            (factor_group, weights) 
            for factor_group, weights in zip(strategy.factor_groups, strategy.default_weights)
        ]
        log.info("使用默认因子权重")
    
    # 使用技术因子进行选股
    for factor_group, weights in strategy.factor_models:
        try:
            # 获取技术因子数据
            scored_stocks = score_stocks_by_technical_factors(initial_list, factor_group, weights, yesterday)
            if scored_stocks:
                final_list.extend(scored_stocks)
                        
        except Exception as e:
            log.error(f"技术因子选股失败: {str(e)}")
            continue
    
    # 去重并限制数量
    final_list = list(set(final_list))[:strategy.stock_num]
    log.info(f"选股完成，共选出 {len(final_list)} 只股票")
    return final_list

def score_stocks_by_technical_factors(stock_list, factor_group, weights, date):
    """根据技术因子给股票打分"""
    try:
        # 获取技术因子数据
        factor_data = {}
        
        for factor in factor_group:
            try:
                # 使用get_factor_values获取技术因子
                values = get_factor_values(securities=stock_list, factors=[factor], 
                                        count=1, end_date=date)
                if values and factor in values and not values[factor].empty:
                    factor_data[factor] = values[factor].iloc[0]
            except:
                continue
        
        if not factor_data:
            return []
        
        # 计算得分
        scores = {}
        for stock in stock_list:
            total_score = 0
            valid_factors = 0
            
            for i, factor in enumerate(factor_group):
                if factor in factor_data and stock in factor_data[factor].index:
                    try:
                        factor_value = factor_data[factor][stock]
                        if not pd.isna(factor_value) and i < len(weights):
                            total_score += weights[i] * factor_value
                            valid_factors += 1
                    except:
                        continue
            
            if valid_factors > 0:
                scores[stock] = total_score
        
        if not scores:
            return []
        
        # 按得分排序
        sorted_stocks = sorted(scores.keys(), key=lambda x: scores[x], reverse=True)
        return sorted_stocks[:max(10, int(0.1 * len(sorted_stocks)))]
        
    except Exception as e:
        log.error(f"技术因子打分失败: {str(e)}")
        return []

# 过滤函数保持不变...
def filter_new_stock(context, stock_list):
    """过滤次新股"""
    if not stock_list:
        return []
    
    yesterday = context.previous_date
    result = []
    for stock in stock_list:
        try:
            start_date = get_security_info(stock).start_date
            if (yesterday - start_date).days > 180:
                result.append(stock)
        except:
            continue
    return result

def filter_paused_stock(stock_list):
    """过滤停牌股票"""
    if not stock_list:
        return []
    
    current_data = get_current_data()
    return [stock for stock in stock_list if not current_data[stock].paused]

def filter_st_stock(stock_list):
    """过滤ST股票"""
    if not stock_list:
        return []
    
    current_data = get_current_data()
    return [stock for stock in stock_list
            if not current_data[stock].is_st
            and 'ST' not in current_data[stock].name
            and '*' not in current_data[stock].name
            and '退' not in current_data[stock].name]

def filter_kcbj_stock(stock_list):
    """过滤科创北交股票"""
    if not stock_list:
        return []
    
    return [stock for stock in stock_list if not stock.startswith(('4', '8', '68'))]

def filter_limitup_stock(context, stock_list):
    """过滤涨停股票"""
    if not stock_list:
        return []
    
    try:
        result = []
        for stock in stock_list:
            if stock in context.portfolio.positions:
                result.append(stock)
            else:
                try:
                    price_data = get_price(stock, end_date=context.current_dt, 
                                         frequency='1m', fields=['close'], count=1)
                    current_data = get_current_data()
                    if (price_data is not None and len(price_data) > 0 and 
                        not np.isnan(price_data['close'].iloc[0]) and 
                        price_data['close'].iloc[0] < current_data[stock].high_limit):
                        result.append(stock)
                except:
                    continue
        return result
    except:
        return stock_list

def filter_limitdown_stock(context, stock_list):
    """过滤跌停股票"""
    if not stock_list:
        return []
    
    try:
        result = []
        for stock in stock_list:
            if stock in context.portfolio.positions:
                result.append(stock)
            else:
                try:
                    price_data = get_price(stock, end_date=context.current_dt, 
                                         frequency='1m', fields=['close'], count=1)
                    current_data = get_current_data()
                    if (price_data is not None and len(price_data) > 0 and 
                        not np.isnan(price_data['close'].iloc[0]) and 
                        price_data['close'].iloc[0] > current_data[stock].low_limit):
                        result.append(stock)
                except:
                    continue
        return result
    except:
        return stock_list

def weekly_adjustment(context):
    """周度调仓"""
    strategy = g.strategy
    if not strategy.no_trading_today_signal:
        target_list = get_stock_list(context)[:strategy.stock_num]
        
        # 调仓卖出
        for stock in strategy.hold_list:
            if stock not in target_list and stock not in strategy.yesterday_HL_list:
                log.info(f"卖出[{stock}]")
                position = context.portfolio.positions[stock]
                close_position(position)
        
        # 调仓买入
        position_count = len(context.portfolio.positions)
        target_num = len(target_list)
        
        if target_num > position_count and context.portfolio.cash > 0:
            value = context.portfolio.cash / (target_num - position_count)
            for stock in target_list:
                if stock not in context.portfolio.positions or context.portfolio.positions[stock].total_amount == 0:
                    if open_position(stock, value):
                        if len(context.portfolio.positions) == target_num:
                            break

def check_limit_up(context):
    """检查涨停股票"""
    strategy = g.strategy
    if strategy.yesterday_HL_list:
        for stock in strategy.yesterday_HL_list:
            try:
                current_data = get_price(stock, end_date=context.current_dt, frequency='1m', 
                                       fields=['close','high_limit'], count=1, panel=False, fill_paused=True)
                if not current_data.empty and current_data.iloc[0,0] < current_data.iloc[0,1]:
                    log.info(f"[{stock}]涨停打开，卖出")
                    position = context.portfolio.positions[stock]
                    close_position(position)
            except:
                continue

def today_is_between(context, start_date, end_date):
    """判断日期是否在范围内"""
    today = context.current_dt.strftime('%m-%d')
    return start_date <= today <= end_date

def close_account(context):
    """清仓"""
    strategy = g.strategy
    if strategy.no_trading_today_signal and strategy.hold_list:
        for stock in strategy.hold_list:
            try:
                position = context.portfolio.positions[stock]
                close_position(position)
                log.info(f"卖出[{stock}]")
            except:
                continue

def open_position(security, value):
    """开仓"""
    try:
        order = order_target_value(security, value)
        return order is not None and order.filled > 0
    except:
        return False

def close_position(position):
    """平仓"""
    try:
        order = order_target_value(position.security, 0)
        return order is not None and order.filled > 0
    except:
        return False

def print_position_info(context):
    """打印持仓信息"""
    total_value = context.portfolio.total_value
    cash = context.portfolio.cash
    positions_count = len(context.portfolio.positions)
    
    strategy = g.strategy
    mode = "防御模式" if strategy.in_defensive_mode else "正常模式"
    
    log.info(f"总资产: {total_value:.2f}, 现金: {cash:.2f}, 持仓数: {positions_count}, 当前模式: {mode}")