行为金融学在量化策略中的应用：捕捉散户情绪与认知偏差

import pandas as pd
import numpy as np

def calculate_retail_sentiment(stock_data, holdings_data):
    """
    计算散户情绪指标
    
    核心逻辑：
    1. 散户持仓占比上升 → 情绪乐观（反向指标）
    2. 散户交易占比上升 → 情绪高涨（接近顶部）
    3. 新增开户数激增 → 散户入场（反转信号）
    
    Parameters:
    -----------
    stock_data : pd.DataFrame
        股票行情数据（包含收盘价为close）
    holdings_data : pd.DataFrame
        持仓数据（包含散户持仓占比retail_ratio）
    
    Returns:
    --------
    sentiment_score : pd.Series
        情绪得分（正值=乐观，负值=悲观）
    """
    # 1. 散户持仓变化
    holdings_change = holdings_data['retail_ratio'].diff()
    
    # 2. 交易热度（成交量/自由流通市值）
    turnover = stock_data['volume'] / stock_data['free_float_shares']
    turnover_zscore = (turnover - turnover.rolling(60).mean()) / turnover.rolling(60).std()
    
    # 3. 融资余额变化（杠杆资金）
    margin_change = holdings_data['margin_balance'].pct_change(5)
    
    # 综合情绪指标（标准化后等权加权）
    sentiment_score = (
        -holdings_change.rolling(20).mean() +  # 散户增持 → 看空
        0.5 * turnover_zscore +                # 换手率飙升 → 看空
        -0.3 * margin_change.rolling(10).mean()  # 杠杆资金激增 → 看空
    )
    
    # 标准化到 [-1, 1]
    sentiment_score = (sentiment_score - sentiment_score.mean()) / sentiment_score.std()
    
    return sentiment_score

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

class SocialMediaSentiment:
    """
    社交媒体情绪分析
    
    数据源：
    - 股吧/雪球/东方财富评论
    - 微博/微信公众号
    - Reddit/Twitter（A股关注度低）
    
    方法：
    - FinBERT（金融领域预训练）
    - 情感词典（Loughran-McDonald）
    - 主题模型（LDA/BERTopic）
    """
    
    def __init__(self, model_name="yiyanghkust/finbert-tone"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model.eval()
    
    def extract_sentiment(self, texts):
        """
        批量提取文本情感
        
        Returns:
        --------
        sentiments : list of dict
            每个文本的情感得分（Positive/Negative/Neutral）
        """
        sentiments = []
        
        for text in texts:
            # Tokenize
            inputs = self.tokenizer(
                text,
                return_tensors="pt",
                truncation=True,
                max_length=512,
                padding=True
            )
            
            # 推理
            with torch.no_grad():
                outputs = self.model(**inputs)
                probabilities = torch.softmax(outputs.logits, dim=1)
            
            # 解析结果
            sentiment = {
                'positive': probabilities[0][0].item(),
                'negative': probabilities[0][1].item(),
                'neutral': probabilities[0][2].item()
            }
            sentiments.append(sentiment)
        
        return sentiments
    
    def aggregate_sentiment(self, sentiments, method='weighted'):
        """
        聚合多条文档的情感
        
        Parameters:
        -----------
        method : str
            'simple' - 简单平均
            'weighted' - 按点赞数/转发数加权
            'time_decay' - 时间衰减加权
        """
        if method == 'simple':
            avg_sentiment = {
                'positive': np.mean([s['positive'] for s in sentiments]),
                'negative': np.mean([s['negative'] for s in sentiments]),
                'neutral': np.mean([s['neutral'] for s in sentiments])
            }
        
        elif method == 'weighted':
            # 假设sentiments中包含权重信息
            weights = np.array([s.get('weight', 1.0) for s in sentiments])
            weights = weights / weights.sum()
            
            avg_sentiment = {
                'positive': np.sum([s['positive'] * w for s, w in zip(sentiments, weights)]),
                'negative': np.sum([s['negative'] * w for s, w in zip(sentiments, weights)]),
                'neutral': np.sum([s['neutral'] * w for s, w in zip(sentiments, weights)])
            }
        
        return avg_sentiment
    
    def construct_trading_signal(self, sentiment_history, threshold=0.7):
        """
        构建交易信号
        
        逻辑：
        - 情绪极度乐观（Positive > threshold）→ 卖出信号
        - 情绪极度悲观（Negative > threshold）→ 买入信号
        """
        signals = pd.Series(0, index=sentiment_history.index)
        
        signals[sentiment_history['positive'] > threshold] = -1  # 看空
        signals[sentiment_history['negative'] > threshold] = 1   # 看多
        
        return signals

def calculate_institutional_herding(fund_holdings, window=20):
    """
    计算机构羊群效应指标
    
    逻辑：
    1. 多只基金同时增持同一只股票 → 羊群效应
    2. 羊群效应越明显 → 未来收益越差（过度反应）
    
    Parameters:
    -----------
    fund_holdings : pd.DataFrame
        基金持仓数据（fund_id, stock_code, date, shares）
    window : int
        滚动窗口长度
    
    Returns:
    --------
    herding_score : pd.Series
        羊群效应得分
    """
    herding_scores = {}
    
    for stock in fund_holdings['stock_code'].unique():
        stock_data = fund_holdings[fund_holdings['stock_code'] == stock]
        
        # 计算每只基金的持仓变化
        stock_data = stock_data.sort_values('date')
        stock_data['shares_change'] = stock_data.groupby('fund_id')['shares'].diff()
        
        # 计算同向交易比例（Herding Measure）
        for date in stock_data['date'].unique():
            date_data = stock_data[stock_data['date'] == date]
            
            buy_count = (date_data['shares_change'] > 0).sum()
            sell_count = (date_data['shares_change'] < 0).sum()
            total = len(date_data)
            
            # H = |p - 0.5|，p为买入比例
            p = buy_count / total if total > 0 else 0.5
            herding = abs(p - 0.5)
            
            herding_scores[(stock, date)] = herding
    
    return pd.Series(herding_scores)

def herding_reversal_strategy(herding_scores, returns, top_quantile=0.1):
    """
    羊群效应反转策略
    
    逻辑：
    羊群效应最严重的股票，未来会反转
    """
    # 按日期分组，计算每日羊群得分的分位数
    signals = pd.Series(0, index=returns.index)
    
    for date in herding_scores.index.get_level_values(1).unique():
        date_scores = herding_scores.loc[:, date]
        threshold = date_scores.quantile(1 - top_quantile)
        
        # 羊群效应最严重的股票 → 做空
        high_herding_stocks = date_scores[date_scores >= threshold].index
        signals.loc[high_herding_stocks] = -1
    
    # 计算策略收益
    strategy_returns = (signals.shift(1) * returns).sum(axis=1)
    
    return strategy_returns

def calculate_northbound_sentiment(northbound_flow, price_data):
    """
    计算北向资金情绪指标
    
    北向资金被视为"聪明钱"，其动向具有信号意义：
    1. 持续净流入 → 外资看多
    2. 大幅净流出 → 外资避险
    3. 流向反转 → 可能变盘
    
    Parameters:
    -----------
    northbound_flow : pd.DataFrame
        北向资金流向数据（date, net_flow, buy, sell）
    price_data : pd.DataFrame
        股票价格数据
    """
    # 1. 北向资金流向的滚动平均（去除噪声）
    flow_ma = northbound_flow['net_flow'].rolling(10).mean()
    flow_zscore = (flow_ma - flow_ma.rolling(60).mean()) / flow_ma.rolling(60).std()
    
    # 2. 北向资金持仓变化（需要持仓数据）
    # position_change = northbound_holdings.groupby('stock').pct_change()
    
    # 3. 背离信号（价格涨但北向流出 → 看空）
    price_momentum = price_data['close'].pct_change(5)
    divergence = (price_momentum > 0) & (flow_ma < 0)
    
    # 构建信号
    signals = pd.Series(0, index=price_data.index)
    signals[flow_zscore > 1] = 1       # 北向大幅流入 → 看多
    signals[flow_zscore < -1] = -1     # 北向大幅流出 → 看空
    signals[divergence] = -1            # 背离信号 → 看空
    
    return signals

class RetailSentimentStrategy:
    """
    散户情绪反转策略
    
    核心逻辑：
    散户情绪极度乐观时卖出，极度悲观时买入
    
    学术支持：
    - Baker and Wurgler (2006): 散户情绪指数量化
    - Brown and Cliff (2004): 情绪是反转指标
    """
    
    def __init__(self, sentiment_window=20, holding_period=5):
        self.sentiment_window = sentiment_window
        self.holding_period = holding_period
    
    def generate_signals(self, sentiment_data, price_data):
        """
        生成交易信号
        
        Parameters:
        -----------
        sentiment_data : pd.DataFrame
            包含情绪指标（retail_sentiment, social_media_sentiment等）
        price_data : pd.DataFrame
            股票价格数据
        
        Returns:
        --------
        signals : pd.DataFrame
            交易信号矩阵（1=买入, -1=卖出, 0=持有）
        """
        signals = pd.DataFrame(0, index=price_data.index, columns=price_data.columns)
        
        for stock in price_data.columns:
            sentiment = sentiment_data[stock]
            
            # 情绪极端值检测（Z-Score > 2）
            sentiment_zscore = (sentiment - sentiment.rolling(60).mean()) / sentiment.rolling(60).std()
            
            # 买入信号：情绪极度悲观
            buy_signal = sentiment_zscore < -2
            
            # 卖出信号：情绪极度乐观
            sell_signal = sentiment_zscore > 2
            
            # 生成信号
            signals.loc[buy_signal, stock] = 1
            signals.loc[sell_signal, stock] = -1
        
        return signals
    
    def backtest(self, signals, returns, transaction_cost=0.001):
        """
        回测策略
        
        Parameters:
        -----------
        signals : pd.DataFrame
            交易信号
        returns : pd.DataFrame
            股票收益率
        transaction_cost : float
            交易成本（双边）
        
        Returns:
        --------
        performance : dict
            策略绩效指标
        """
        # 计算策略收益
        strategy_returns = (signals.shift(1) * returns).sum(axis=1)
        
        # 扣除交易成本
        turnover = signals.diff().abs().sum(axis=1)
        cost = turnover * transaction_cost
        net_returns = strategy_returns - cost
        
        # 计算绩效指标
        cumulative_returns = (1 + net_returns).cumprod()
        annual_return = net_returns.mean() * 252
        annual_vol = net_returns.std() * np.sqrt(252)
        sharpe = annual_return / annual_vol
        max_drawdown = (cumulative_returns / cumulative_returns.cummax() - 1).min()
        
        performance = {
            'annual_return': annual_return,
            'annual_volatility': annual_vol,
            'sharpe_ratio': sharpe,
            'max_drawdown': max_drawdown,
            'cumulative_returns': cumulative_returns
        }
        
        return performance

def analyst_revision_strategy(analyst_data, returns, window=10):
    """
    分析师修正策略
    
    行为偏差：
    分析师倾向于过度反应（Overreaction）
    - 业绩好 → 大幅上调目标价 → 股价超涨 → 未来反转
    - 业绩差 → 大幅下调目标价 → 股价超跌 → 未来反弹
    
    策略逻辑：
    1. 统计分析师评级上调/下调的数量
    2. 过度上调 → 做空
    3. 过度下调 → 做多
    """
    # 1. 计算评级变化
    analyst_data = analyst_data.sort_values(['stock', 'date'])
    analyst_data['rating_change'] = analyst_data.groupby('stock')['rating'].diff()
    
    # 2. 汇总每日上调/下调数量
    upgrade_count = (analyst_data['rating_change'] > 0).groupby(analyst_data['date']).sum()
    downgrade_count = (analyst_data['rating_change'] < 0).groupby(analyst_data['date']).sum()
    
    # 3. 构建情绪极端值指标
    net_upgrade = upgrade_count - downgrade_count
    net_upgrade_zscore = (net_upgrade - net_upgrade.rolling(60).mean()) / net_upgrade.rolling(60).std()
    
    # 4. 生成信号
    signals = pd.Series(0, index=returns.index)
    signals[net_upgrade_zscore > 2] = -1   # 过度乐观 → 做空
    signals[net_upgrade_zscore < -2] = 1    # 过度悲观 → 做多
    
    # 5. 回测
    strategy_returns = signals.shift(1) * returns
    
    return strategy_returns

def disposition_effect_strategy(price_data, high_volume_stocks):
    """
    处置效应策略
    
    行为偏差：
    投资者倾向于：
    - 过早卖出盈利股票（落袋为安）
    - 过久持有亏损股票（死扛）
    
    价格影响：
    - 盈利股票：卖压 → 短期承压 → 未来上涨（动量）
    - 亏损股票：缺乏卖压 → 短期抗跌 → 未来下跌（反转）
    
    策略：
    买入近期大涨但未涨停的股票（有卖压），卖出近期大跌但未跌停的股票
    """
    signals = pd.DataFrame(0, index=price_data.index, columns=price_data.columns)
    
    for date in price_data.index[20:]:  # 从第20天开始
        # 计算过去N日收益率
        past_returns = price_data.loc[:date].iloc[-20:].pct_change().sum()
        
        # 识别"盈利股票"（近期上涨但未涨停）
        gainers = past_returns[past_returns > 0.05]  # 涨幅>5%
        gainers = gainers[gainers < 0.09]           # 但未涨停（排除涨停股）
        
        # 识别"亏损股票"（近期下跌但未跌停）
        losers = past_returns[past_returns < -0.05]  # 跌幅>5%
        losers = losers[losers > -0.09]              # 但未跌停
        
        # 生成信号
        signals.loc[date, gainers.index] = -1   # 卖出盈利股票（有卖压）
        signals.loc[date, losers.index] = 1     # 买入亏损股票（缺乏卖压）
    
    return signals

def construct_sentiment_factor(sentiment_data, returns, holding_period=5):
    """
    构建情绪因子
    
    方法：
    1. 每月末根据情绪指标打分
    2. 做多情绪最低的股票（极度悲观）
    3. 做空情绪最高的股票（极度乐观）
    4. 持有N日后平仓
    
    Parameters:
    -----------
    sentiment_data : pd.DataFrame
        情绪指标数据（date × stock）
    returns : pd.DataFrame
        股票收益率数据
    holding_period : int
        持仓周期
    
    Returns:
    --------
    factor_returns : pd.Series
        因子收益率序列
    """
    factor_returns = pd.Series(0, index=returns.index)
    
    # 每月末重新平衡
    rebalance_dates = returns.resample('M').last().index
    
    for i, rebalance_date in enumerate(rebalance_dates[:-1]):
        # 获取调仓日情绪数据
        sentiment_scores = sentiment_data.loc[rebalance_date]
        
        # 分组（十分位数）
        quantiles = pd.qcut(sentiment_scores, 10, labels=False)
        
        # 做多最低情绪组（Quantile 0），做空最高情绪组（Quantile 9）
        long_stocks = quantiles[quantiles == 0].index
        short_stocks = quantiles[quantiles == 9].index
        
        # 计算持仓期间收益
        start_idx = returns.index.get_loc(rebalance_date) + 1
        end_idx = min(start_idx + holding_period, len(returns))
        
        for t in range(start_idx, end_idx):
            long_ret = returns.iloc[t][long_stocks].mean()
            short_ret = -returns.iloc[t][short_stocks].mean()  # 做空收益 = -收益率
            factor_returns.iloc[t] = (long_ret + short_ret) / 2
    
    return factor_returns

def construct_herding_factor(herding_scores, returns, holding_period=5):
    """
    构建羊群因子
    
    逻辑：
    羊群效应越严重的股票，未来表现越差（过度反应后的价格反转）
    
    组合构建：
    - 做多羊群效应最低的股票（独立思考）
    - 做空羊群效应最高的股票（盲目跟风）
    """
    factor_returns = pd.Series(0, index=returns.index)
    
    rebalance_dates = returns.resample('M').last().index
    
    for i, rebalance_date in enumerate(rebalance_dates[:-1]):
        # 获取羊群得分
        scores = herding_scores.loc[rebalance_date]
        
        # 分组
        quantiles = pd.qcut(scores, 10, labels=False)
        
        # 做多低羊群组，做空高羊群组
        long_stocks = quantiles[quantiles == 0].index
        short_stocks = quantiles[quantiles == 9].index
        
        # 计算收益
        start_idx = returns.index.get_loc(rebalance_date) + 1
        end_idx = min(start_idx + holding_period, len(returns))
        
        for t in range(start_idx, end_idx):
            long_ret = returns.iloc[t][long_stocks].mean()
            short_ret = -returns.iloc[t][short_stocks].mean()
            factor_returns.iloc[t] = (long_ret + short_ret) / 2
    
    return factor_returns

def construct_overreaction_factor(returns, holding_period=5):
    """
    构建过度反应因子
    
    逻辑：
    过去N日涨幅最大的股票，未来会反转（过度反应）
    过去N日跌幅最大的股票，未来会反弹（反应不足）
    
    这是经典的"反转因子"（Reversal Factor）
    """
    factor_returns = pd.Series(0, index=returns.index)
    
    rebalance_dates = returns.resample('M').last().index
    
    for i, rebalance_date in enumerate(rebalance_dates[:-1]):
        # 计算过去60日收益率
        past_returns = returns.loc[:rebalance_date].iloc[-60:].sum()
        
        # 分组
        quantiles = pd.qcut(past_returns, 10, labels=False)
        
        # 做多过去跌幅最大的（Quantile 0），做空过去涨幅最大的（Quantile 9）
        long_stocks = quantiles[quantiles == 0].index
        short_stocks = quantiles[quantiles == 9].index
        
        # 计算收益
        start_idx = returns.index.get_loc(rebalance_date) + 1
        end_idx = min(start_idx + holding_period, len(returns))
        
        for t in range(start_idx, end_idx):
            long_ret = returns.iloc[t][long_stocks].mean()
            short_ret = -returns.iloc[t][short_stocks].mean()
            factor_returns.iloc[t] = (long_ret + short_ret) / 2
    
    return factor_returns

def behavioral_risk_budget(cov_matrix, sentiment_scores, risk_aversion=2.0):
    """
    行为风险预算模型
    
    改进：
    传统风险平价假设投资者理性
    本模型引入"情绪调整的风险溢价"
    
    逻辑：
    1. 情绪高涨时 → 风险溢价降低 → 降低仓位
    2. 情绪低迷时 → 风险溢价升高 → 增加仓位
    """
    n_assets = cov_matrix.shape[0]
    
    # 情绪调整的风险溢价
    risk_premium = np.ones(n_assets) * 0.05  # 基础风险溢价 5%
    sentiment_adjustment = -0.02 * sentiment_scores  # 情绪越高，溢价越低
    adjusted_premium = risk_premium + sentiment_adjustment
    
    # 优化目标：最大化效用函数 U = E[r] - 0.5 * risk_aversion * σ²
    def objective(weights):
        portfolio_return = np.dot(weights, adjusted_premium)
        portfolio_risk = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
        utility = portfolio_return - 0.5 * risk_aversion * portfolio_risk ** 2
        return -utility  # 负号因为scipy是最小化
    
    # 约束条件
    cons = [{'type': 'eq', 'fun': lambda w: np.sum(w) - 1.0}]
    bounds = [(0, 1) for _ in range(n_assets)]
    
    # 优化
    result = minimize(
        objective,
        np.ones(n_assets) / n_assets,
        method='SLSQP',
        constraints=cons,
        bounds=bounds
    )
    
    return result.x

# 回测设置
initial_capital = 1000000
transaction_cost = 0.001  # 双边0.1%
holding_period = 5

# 策略1：散户情绪反转
strategy1_returns = RetailSentimentStrategy().backtest(signals, returns, transaction_cost)

# 策略2：分析师修正
strategy2_returns = analyst_revision_strategy(analyst_data, returns)

# 策略3：因子组合（等权）
factor_returns = (sentiment_factor + herding_factor + overreaction_factor) / 3
strategy3_returns = factor_returns

# 绩效对比
performance_comparison = pd.DataFrame({
    '散户情绪反转': calculate_performance(strategy1_returns),
    '分析师修正': calculate_performance(strategy2_returns),
    '因子组合': calculate_performance(strategy3_returns)
})

print(performance_comparison)