星耀数智因子分析skill

pip install tgw>=1.0.8.7
pip install AmazingData>=1.1.15

# Windows CMD
set AD_USERNAME=your_username
set AD_PASSWORD=your_password
set AD_HOST=101.230.159.235
set AD_PORT=8600

# Windows PowerShell
$env:AD_USERNAME="your_username"
$env:AD_PASSWORD="your_password"
$env:AD_HOST="101.230.159.235"
$env:AD_PORT="8600"

from scripts.data_provider import DataProvider

dp = DataProvider()
stock_list = dp.get_stock_list()[:200]  # 取前200只
close = dp.get_close_price(stock_list, begin_date, end_date)
benchmark_df = dp.get_benchmark('000300.SH', begin_date, end_date)

ma5 = close.rolling(5).mean()
factor_raw = (close - ma5) / ma5

from AmazingData.factor_analysis import FactorPreProcessing, ExtremeMethod, ScaleMethod, FillNanMethod

fpp = FactorPreProcessing(factor_raw)
fpp.extreme_processing({ExtremeMethod.MAD.value: {'median_multiple': 5}})
fpp.scale_processing(ScaleMethod.Z_SCORE.value)
fpp.fill_nan_processing(FillNanMethod.MEDIAN.value)
factor = fpp.processed_data

from AmazingData.factor_analysis import (
    IcAnalysis, RegressionAnalysis, StratificationAnalysis, FactorCrowdingAnalysis
)

# IC 分析
ia = IcAnalysis(factor, factor_name, close, ic_decay=20)
ia.cal_ic_df(method='spearmanr')
ia.cal_ic_indicator()

# 回归法分析
ra = RegressionAnalysis(factor, factor_name, close, benchmark_df)
ra.cal_factor_return()
ra.cal_t_value_statistics()
ra.cal_net_analysis()
ra.cal_acf(nlags=10)

# 分层法分析
bm_series = benchmark_df['close'] / benchmark_df['close'].iloc[0]
sa = StratificationAnalysis(factor, close, group_num=5, ascending=False, benchmark=bm_series)
sa.run()
sa._backtest.calc_signal_decay_reversal(10)

# 因子拥挤度
mc = pd.DataFrame(...)  # 流通市值
fca = FactorCrowdingAnalysis(factor, close, mc, group_num=5, ascending=False)
fca.calc_all(window=60)

from scripts.report_renderer import FactorAnalysisReport

report = FactorAnalysisReport(factor_name, title)
report.add_definition_section(html_desc)      # 因子定义
report.add_ic_section(ia.ic_df, ia.ic_result, ia.p_value_df)
report.add_regression_section(ra.factor_return, ra.factor_t_value, ra.net_analysis_result)
report.add_stratification_section(sa.group_navs, sa.group_metrics, sa.turnover,
                                   sa.signal_decay, sa.signal_reversal, sa.long_short_nav)
report.add_crowding_section(fca.crowding_summary(), crowding_series)
report.generate(output_path)

# 假设有 3 个因子：ma5 偏离度、ma20 偏离度、换手率因子
ma5 = close.rolling(5).mean()
ma20 = close.rolling(20).mean()
factor_ma5 = (close - ma5) / ma5
factor_ma20 = (close - ma20) / ma20
factor_turnover = dp.get_kline(stock_list, begin_date, end_date, fields=['volume'])
factor_turnover = factor_turnover.droplevel(0, axis=1)  # 去掉 MultiIndex 层级

# 组装为字典
factors = {
    'ma5_deviation': factor_ma5,
    'ma20_deviation': factor_ma20,
    'volume': factor_turnover,
}

from scripts.run_analysis import run_multi_factor_analysis

run_multi_factor_analysis(
    factors=factors,                              # {因子名: 因子值 DataFrame}
    factor_names=['ma5', 'ma20', 'volume'],       # 因子名列表（决定合成顺序）
    close_price=close,
    benchmark_df=benchmark_df,
    market_cap=mc,
    output_path='multi_factor_report.html',
    group_num=5,
    ic_decay=20,
    weight_method='ic_mean',                      # 加权方法（见下方说明）
    use_orthogonal=True,                          # 是否进行正交化
)

from AmazingData.factor_analysis import (
    FactorPreProcessing, ExtremeMethod, ScaleMethod, FillNanMethod,
    CollinearityAnalysis, FactorOrthogonalization, FactorWeighting,
    StockScorer,
)

# Step 1: 逐因子预处理
processed_factors = {}
for name, factor_raw in factors.items():
    fpp = FactorPreProcessing(factor_raw)
    fpp.extreme_processing({ExtremeMethod.MAD.value: {'median_multiple': 5}})
    fpp.scale_processing(ScaleMethod.Z_SCORE.value)
    fpp.fill_nan_processing(FillNanMethod.MEDIAN.value)
    processed_factors[name] = fpp.processed_data

# Step 2: 共线性检测
ca = CollinearityAnalysis(processed_factors)
ca.cal_correlation()          # 因子间相关系数矩阵
ca.cal_vif()                  # 方差膨胀因子 VIF
ca.cal_condition_number()     # 条件数
print(f"VIF: {ca.vif_df}")
print(f"条件数: {ca.condition_number}")

# Step 3: 因子正交化（消除共线性）
orthogonalized = processed_factors  # 默认使用原始因子
if ca.condition_number > 30 or any(ca.vif_df['VIF'] > 10):
    fa = FactorOrthogonalization(processed_factors)
    fa.orthogonalize(method='symmetric')  # symmetric(推荐)/schmidt/canonical
    orthogonalized = fa.orthogonalized_factors

# Step 4: 因子加权
fw = FactorWeighting()
fw.cal_weight(orthogonalized, method='ic_mean')  # 8 种方法可选
weights = fw.weights  # Dict[str, float]

# Step 5: 个股打分 → 合成因子
ss = StockScorer(orthogonalized, weights)
ss.calculate()
composite_factor = ss.composite_score  # pd.DataFrame (index=日期, columns=股票代码)

# Step 6: 对合成因子做单因子分析
from scripts.run_analysis import run_factor_analysis

run_factor_analysis(
    factor_raw=composite_factor,
    factor_name='multi_factor_composite',
    close_price=close,
    benchmark_df=benchmark_df,
    market_cap=mc,
    output_path='composite_report.html',
)

# 在 references/factor_analysis_algorithms.md 中搜索关键字：
grep -n "VIF|条件数|medcouple|Ledoit|FSC|WLS|IC_IR|单调性" references/factor_analysis_algorithms.md