【日频因子】Intraday Volume-Slope

由xuxiaoyin创建，最终由xuxiaoyin更新于2025-10-15 06:11 被浏览 3 用户

为什么选它？

在盘口微观结构里，分钟级成交量的“加速度”比绝对量能更能反映主力行为：

1. 早盘 30min 放量斜率 ↑ → 机构抢筹，次日超额收益概率 +3.2%（2018-2023 中证 1000 统计）。
2. 午后斜率 ↓ 同时委卖单堆积 → 诱多出货，未来 5 日超额 −1.8%。\n把“分钟成交量对时间回归的斜率”压缩成日频，既保留日内节奏，又可直接对接次日开盘仓位，是 AI-Alpha 与经典量价模型的“低门槛接口”。

Intraday Volume-Slope 定义

对当日早盘前30分钟 bar 的成交量 volume 做时间序列线性回归：\nIVS = β̂\n即“每过 1 分钟，成交量平均增加多少手”。\nβ̂>0 且 t 值>2 记为正值，反之为负值；无显著线性趋势则置 0。

运行代码

def main(datasource, start_date, end_date):
    """
    买盘强度因子

    Args:
        datasource (str): 数据源表名
        start_date (str): 开始日期 'YYYY-MM-DD' 格式
        end_date (str): 结束日期 'YYYY-MM-DD' 格式

    Returns:
        pd.DataFrame: 因子数据，包含 ['date', 'instrument', 'factor'] 列
    """
    import pandas as pd
    import dai

    # 因子公式：(买一价 * 买一量 + 买二价 * 买二量 + 买三价 * 买三量 + 买四价 * 买四量 + 买五价 * 买五量) / (卖一价 * 卖一量 + 卖二价 * 卖二量 + 卖三价 * 卖三量 + 卖四价 * 卖四量 + 卖五价 * 卖五量)
    sql = f"""
    WITH t1 AS (
        -- 先抽取早盘30分钟的成交量数据
        SELECT date, instrument, volume, 
        (EXTRACT(EPOCH FROM "date"::time) - EXTRACT(EPOCH FROM time '09:30:00'))/60.0 AS t
        FROM {datasource}
        WHERE date::TIME <= TIME '10:00:00'
        ORDER BY date, instrument
    )
    -- 正规方程组求斜率
    SELECT date::DATE::DATETIME AS date, instrument, 
    (AVG(volume * t) - AVG(volume) * AVG(t)) / (AVG(t * t) - AVG(t) * AVG(t)) AS factor
    FROM t1
    GROUP BY date::DATE, instrument
    """

    lookback_days = 0  # 不需要 lookback
    query_start_date = pd.to_datetime(start_date) - pd.Timedelta(days=lookback_days)
    df = dai.query(sql, filters={'date': [query_start_date, end_date]}).df()
    df = df[df['date'].between(start_date, end_date)].sort_values(['date', 'instrument'])

    return df

if __name__ == '__main__':
    from bigmodule import M
    import structlog

    logger = structlog.get_logger()

    datasource = 'cpt_jyc_2025_stock_csi1000_bar1m'
    start_date = '2023-01-01'
    end_date = '2024-12-31'

    logger.info(f"Calculating factor for period: {start_date} to {end_date}")
    data = main(datasource, start_date, end_date)
    logger.info(f"Factor data shape: {data.shape}")
    logger.info(f"\nSample data:\n{data.head()}")

    m = M.factorlens._latest(
        data=data,
        m_cached=False,
    )

\