今天继续安更新计划更新分析内容，今天分享是东方证券的研报基于量价关系度量股票的买卖压力，研报只做学习参考不做商业用途

1 策略算法原理2qmt使用教程3策略算法原理4量化研报5 ptrad使用教程6 因子分析7策略算法原理

这份由东方证券于2019年10月29日发布的研究报告，核心贡献在于

系统性地构建了一套基于量价关系来度量股票买卖压力的指标体系（APB指标）

，并将其应用于多时间尺度的选股因子构建。研究发现，这些因子在全市场及不同样本空间中均具备显著的选股能力，且能提供与传统因子互补的增量Alpha信息。

二、 核心方法论：APB指标与买卖压力因子

报告也客观地指出了应用时需要注意的风险，这与知识库中其他研报的共识一致：

将此报告置于您提供的整个知识库背景下，可以看出其独特价值和关联：

逻辑清晰、实证扎实、实用性强的优秀量价因子研究

。它没有停留在简单的价量结合，而是创新性地提出了“买卖压力”的量化思想，并通过多时间尺度分解，构建了一个立体化的因子体系。其核心亮点在于证明了

日内与日间价量信息低相关且可互补

，这为构建更加鲁棒的多因子选股模型提供了宝贵的思路。对于量化研究者而言，其方法论（APB指标构建、多尺度分析、严谨的稳健性检验）具有很高的借鉴价值。

下面是利用模拟数据算法生成的核心思路参考，代码只是参考学习思路，不做投资参考

# -*- coding: utf-8 -*-
"""
东方证券APB买卖压力因子计算
参考：东方证券《因子选股系列研究六十：基于量价关系度量股票的买卖压力》
核心指标：APB = log(等权加权均价 / 成交量加权均价)
三个时间尺度：
    APB_1m: 月度周期计算
    APB_5d: 基于过去5个交易日滚动计算，过去1个月求均值
    APB_1d: 基于日内5分钟K线计算，过去1个月求均值
"""
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')
# ==================== 第一部分：数据准备（模拟数据） ====================
def generate_sample_data(start_date='2025-01-01', end_date='2025-03-31', n_stocks=100):
    """
    生成模拟的股票日频和分钟频数据，用于演示APB因子计算。
    实战中应替换为真实的行情数据（如来自Wind、Tushare等）。
    """
    np.random.seed(42)
    
    # 生成交易日序列
    all_dates = pd.date_range(start_date, end_date, freq='B')
    stock_codes = [f'STK{i:03d}' for i in range(n_stocks)]
    
    # ---------- 日频数据生成 ----------
    daily_data = []
    for date in all_dates:
        for code in stock_codes:
            # 模拟基础价格（假设服从随机游走）
            base_price = 10 + np.random.randn() * 2
            
            # 模拟日频VWAP（成交量加权平均价）和成交量
            vwap_daily = base_price * (1 + np.random.randn() * 0.02)
            volume_daily = np.random.lognormal(12, 1.5)  # 成交量
            
            daily_data.append({
                'date': date,
                'code': code,
                'vwap': vwap_daily,
                'volume': volume_daily
            })
    
    df_daily = pd.DataFrame(daily_data)
    
    # ---------- 分钟频数据生成（用于APB_1d） ----------
    # 假设每天有240分钟（4小时交易）
    minute_data = []
    for date in all_dates:
        for code in stock_codes:
            # 生成日内分钟价格序列（假设有趋势和波动）
            intraday_trend = np.linspace(-0.01, 0.01, 240)
            intraday_noise = np.random.randn(240) * 0.005
            minute_prices = 10 * (1 + intraday_trend + intraday_noise)
            
            # 生成分钟成交量（通常有U形分布）
            time_index = np.arange(240)
            volume_pattern = 0.5 + 0.5 * np.sin(2 * np.pi * time_index / 240 - np.pi/2)  # U形
            minute_volumes = np.random.lognormal(8, 1) * volume_pattern
            
            for t in range(240):
                minute_data.append({
                    'datetime': date + timedelta(minutes=t),
                    'date': date,
                    'code': code,
                    'price': minute_prices[t],
                    'volume': minute_volumes[t]
                })
    
    df_minute = pd.DataFrame(minute_data)
    
    return df_daily, df_minute
# ==================== 第二部分：核心APB指标计算函数 ====================
def calculate_apb(vwap_array, volume_array):
    """
    计算单周期APB指标
    APB = log(等权加权均价 / 成交量加权均价)
    参数:
        vwap_array: 数组，某周期内各时间点的VWAP价格
        volume_array: 数组，对应时间点的成交量
    返回:
        apb值 (float)
    """
    # 等权加权均价（算术平均）
    equal_weighted_avg = np.mean(vwap_array)
    
    # 成交量加权均价（VWAP of the period）
    if np.sum(volume_array) > 0:
        volume_weighted_avg = np.sum(vwap_array * volume_array) / np.sum(volume_array)
    else:
        volume_weighted_avg = equal_weighted_avg
    
    # 避免除零或负数
    if volume_weighted_avg <= 0:
        return 0.0
    
    # 计算APB（取对数）
    apb = np.log(equal_weighted_avg / volume_weighted_avg)
    
    return apb
# ==================== 第三部分：三个时间尺度APB因子计算 ====================
def calculate_apb_1m(df_daily, lookback_month=1):
    """
    计算月度APB因子 (APB_1m)
    步骤：
    1. 对每只股票，按月计算APB（使用当月所有交易日的vwap和volume）
    2. 将月度APB值赋给该月最后一个交易日
    """
    df = df_daily.copy()
    df['year_month'] = df['date'].dt.to_period('M')
    
    # 按股票和月份分组计算APB
    def calc_monthly_apb(group):
        if len(group) < 5:  # 至少需要5个交易日
            return np.nan
        return calculate_apb(group['vwap'].values, group['volume'].values)
    
    monthly_apb = df.groupby(['code', 'year_month']).apply(calc_monthly_apb)
    monthly_apb = monthly_apb.reset_index(name='apb_1m_raw')
    
    # 找到每月的最后一个交易日
    last_dates = df.groupby(['code', 'year_month'])['date'].max().reset_index()
    monthly_data = pd.merge(last_dates, monthly_apb, on=['code', 'year_month'])
    
    # 合并回原始数据（向前填充，直到下一个月度值出现）
    df_result = pd.merge(df, monthly_data[['code', 'date', 'apb_1m_raw']], 
                         on=['code', 'date'], how='left')
    df_result['apb_1m_raw'] = df_result.groupby('code')['apb_1m_raw'].ffill()
    
    # 过去N个月均值（研报中是过去1个月，这里lookback_month可调）
    df_result['apb_1m'] = df_result.groupby('code')['apb_1m_raw'].rolling(
        window=lookback_month, min_periods=1).mean().reset_index(level=0, drop=True)
    
    return df_result
def calculate_apb_5d(df_daily, lookback_days=5, smooth_window=20):
    """
    计算5日滚动APB因子 (APB_5d)
    步骤：
    1. 对每只股票，滚动计算过去5个交易日的APB
    2. 对APB序列取过去1个月（约20个交易日）的均值
    """
    df = df_daily.copy()
    
    # 按股票分组，滚动计算5日APB
    def rolling_5d_apb(stock_df):
        vwap = stock_df['vwap'].values
        volume = stock_df['volume'].values
        dates = stock_df['date'].values
        
        apb_5d_raw = np.full(len(stock_df), np.nan)
        
        for i in range(lookback_days-1, len(stock_df)):
            start_idx = i - lookback_days + 1
            vwap_window = vwap[start_idx:i+1]
            volume_window = volume[start_idx:i+1]
            
            if len(vwap_window) == lookback_days:
                apb_5d_raw[i] = calculate_apb(vwap_window, volume_window)
        
        return apb_5d_raw
    
    df['apb_5d_raw'] = df.groupby('code', group_keys=False).apply(
        lambda x: pd.Series(rolling_5d_apb(x), index=x.index))
    
    # 过去1个月平滑（约20个交易日）
    df['apb_5d'] = df.groupby('code')['apb_5d_raw'].rolling(
        window=smooth_window, min_periods=1).mean().reset_index(level=0, drop=True)
    
    return df
def calculate_apb_1d(df_daily, df_minute, smooth_window=20):
    """
    计算日内APB因子 (APB_1d)
    步骤：
    1. 对每只股票每个交易日，基于日内分钟数据计算APB
    2. 对日度APB序列取过去1个月（约20个交易日）的均值
    """
    # 从分钟数据计算每日的日内APB
    def calc_daily_intraday_apb(date_group):
        if len(date_group) < 10:  # 至少需要一定数量的分钟数据
            return np.nan
        
        # 使用分钟级的price和volume计算APB
        return calculate_apb(date_group['price'].values, date_group['volume'].values)
    
    # 计算每只股票每天的日内APB
    daily_apb = df_minute.groupby(['code', 'date']).apply(calc_daily_intraday_apb)
    daily_apb = daily_apb.reset_index(name='apb_1d_daily')
    
    # 合并到日频数据
    df = df_daily.copy()
    df = pd.merge(df, daily_apb, on=['code', 'date'], how='left')
    
    # 过去1个月平滑（约20个交易日）
    df['apb_1d'] = df.groupby('code')['apb_1d_daily'].rolling(
        window=smooth_window, min_periods=1).mean().reset_index(level=0, drop=True)
    
    return df
# ==================== 第四部分：因子合并与中性化处理 ====================
def merge_all_factors(df_1m, df_5d, df_1d):
    """
    合并三个时间尺度的APB因子到一个DataFrame
    """
    # 提取需要的列
    factors_1m = df_1m[['date', 'code', 'apb_1m']].copy()
    factors_5d = df_5d[['date', 'code', 'apb_5d']].copy()
    factors_1d = df_1d[['date', 'code', 'apb_1d']].copy()
    
    # 合并
    merged = pd.merge(factors_1m, factors_5d, on=['date', 'code'], how='outer')
    merged = pd.merge(merged, factors_1d, on=['date', 'code'], how='outer')
    
    return merged
def neutralize_factors(df_factors, df_industry_market_cap):
    """
    对因子进行行业市值中性化处理（如研报中所做）
    参数:
        df_factors: 包含因子的DataFrame
        df_industry_market_cap: 包含行业和市值信息的DataFrame
    返回:
        中性化后的因子DataFrame
    """
    # 这里简化处理：假设我们已经有了行业和市值数据
    # 实际应用中需要从外部获取
    df = pd.merge(df_factors, df_industry_market_cap, on=['date', 'code'], how='left')
    
    neutralized_factors = df_factors.copy()
    
    for factor_col in ['apb_1m', 'apb_5d', 'apb_1d']:
        if factor_col in df.columns:
            # 按日期进行横截面回归，残差作为中性化因子
            neutralized_col = f'{factor_col}_neutral'
            neutralized_factors[neutralized_col] = np.nan
            
            for date, group in df.groupby('date'):
                if factor_col in group.columns and 'industry' in group.columns and 'market_cap' in group.columns:
                    # 简单示例：先对行业取哑变量，再对市值回归
                    # 实际应使用更严谨的方法
                    industry_dummies = pd.get_dummies(group['industry'], prefix='ind')
                    X = pd.concat([industry_dummies, group['market_cap'].values.reshape(-1, 1)], axis=1)
                    y = group[factor_col].values
                    
                    if len(y) > X.shape[1] + 5:  # 足够样本
                        from sklearn.linear_model import LinearRegression
                        model = LinearRegression()
                        model.fit(X.fillna(0), y)
                        residuals = y - model.predict(X.fillna(0))
                        
                        # 标准化残差
                        if np.std(residuals) > 0:
                            residuals = (residuals - np.mean(residuals)) / np.std(residuals)
                        
                        idx = group.index
                        neutralized_factors.loc[idx, neutralized_col] = residuals
    
    return neutralized_factors
# ==================== 第五部分：主执行流程 ====================
def main():
    print("=" * 80)
    print("东方证券APB买卖压力因子计算")
    print("=" * 80)
    
    # 1. 生成模拟数据
    print("\n[1/4] 生成模拟数据...")
    df_daily, df_minute = generate_sample_data(
        start_date='2025-01-01',
        end_date='2025-03-31',
        n_stocks=50
    )
    print(f"日频数据形状: {df_daily.shape}")
    print(f"分钟数据形状: {df_minute.shape}")
    print(f"时间范围: {df_daily['date'].min()} 到 {df_daily['date'].max()}")
    print(f"股票数量: {df_daily['code'].nunique()}")
    
    # 2. 计算三个时间尺度的APB因子
    print("\n[2/4] 计算APB_1m（月度因子）...")
    df_apb_1m = calculate_apb_1m(df_daily, lookback_month=1)
    
    print("计算APB_5d（5日滚动因子）...")
    df_apb_5d = calculate_apb_5d(df_daily, lookback_days=5, smooth_window=20)
    
    print("计算APB_1d（日内因子）...")
    df_apb_1d = calculate_apb_1d(df_daily, df_minute, smooth_window=20)
    
    # 3. 合并因子
    print("\n[3/4] 合并所有因子...")
    df_factors = merge_all_factors(df_apb_1m, df_apb_5d, df_apb_1d)
    
    # 显示示例
    print("\n因子数据示例（前5行）：")
    print(df_factors.head())
    
    # 4. 基本统计分析（模拟研报中的分析）
    print("\n[4/4] 因子基本统计分析...")
    
    # 因子描述性统计
    print("\n因子描述性统计：")
    stats = df_factors[['apb_1m', 'apb_5d', 'apb_1d']].describe()
    print(stats)
    
    # 因子间相关性（模拟研报表3.2）
    print("\n因子间相关系数（Spearman秩相关）：")
    corr_matrix = df_factors[['apb_1m', 'apb_5d', 'apb_1d']].corr(method='spearman')
    print(corr_matrix)
    
    # 检查因子分布（模拟研报图2）
    print("\n因子值小于0的比例（类似处置效应检查）：")
    for col in ['apb_1m', 'apb_5d', 'apb_1d']:
        if col in df_factors.columns:
            pct_negative = (df_factors[col] < 0).mean() * 100
            print(f"{col}: {pct_negative:.1f}% 的值为负")
    
    # 因子时间序列示例（取一只股票）
    sample_stock = df_factors['code'].iloc
    sample_data = df_factors[df_factors['code'] == sample_stock].sort_values('date')
    
    print(f"\n示例股票 {sample_stock} 的因子时间序列（前10个交易日）：")
    print(sample_data[['date', 'apb_1m', 'apb_5d', 'apb_1d']].head(10))
    
    return df_factors, df_daily, df_minute
# ==================== 执行主函数 ====================
if __name__ == "__main__":
    # 运行完整计算流程
    df_factors, df_daily, df_minute = main()
    
    # 提供后续分析建议
    print("\n" + "=" * 80)
    print("后续分析建议")
    print("=" * 80)
    print("""
# 1. 因子选股能力回测（模拟研报第2章）
# 需要准备未来收益率数据
# future_ret = df_daily.groupby('code')['vwap'].pct_change().shift(-1)  # 示例
# 2. 因子分组收益（模拟研报图4-6）
def factor_group_backtest(factor_values, future_returns, n_groups=10):
    \"\"\"因子十分组回测\"\"\"
    groups = pd.qcut(factor_values, q=n_groups, labels=False, duplicates='drop')
    group_returns = future_returns.groupby(groups).mean()
    return group_returns
# 3. 因子IC分析（模拟研报表2.2）
def calculate_rank_ic(factor_series, ret_series):
    \"\"\"计算Rank IC\"\"\"
    from scipy.stats import spearmanr
    valid_mask = ~factor_series.isna() & ~ret_series.isna()
    if sum(valid_mask) > 10:
        ic, _ = spearmanr(factor_series[valid_mask], ret_series[valid_mask])
        return ic
    return np.nan
# 4. 因子中性化（模拟研报第3章）
# 需要行业和市值数据
# df_neutral = neutralize_factors(df_factors, df_industry_market_cap)
    """)

不懂的问我就可以，核心学习参考

量化研报 | 量价关系怎么度量买卖压力？一篇讲透

投资顾问

算法

大数据

东方证券研报创新构建APB买卖压力指标，通过量化单位成交量推动价格变化的效率，从月度、5日、日内三个时间尺度捕捉买卖力量对比。实证显示各周期因子选股能力显著，买压强的股票未来一个月跑赢卖压股，且与传统因子低相关，提供独立Alpha增量，兼具稳健性与大市值适用性。

2026上云采购 | AI焕新·智启新局

lexiang

腾讯云BI特惠9.9元起！


文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

量化研报 | 量价关系怎么度量买卖压力？一篇讲透

量化研报 | 量价关系怎么度量买卖压力？一篇讲透

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐