新闻情感分析驱动的事件驱动策略：从NLP到超额收益

import pandas as pd
import requests
from datetime import datetime, timedelta
import json

class NewsCollector:
    """新闻数据采集器"""
    
    def __init__(self, api_key=None):
        self.api_key = api_key
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
    
    def collect_sina_finance(self, stock_code, start_date, end_date):
        """
        采集新浪财经新闻
        
        Parameters:
        -----------
        stock_code: str, 股票代码（如'600000'）
        start_date: str, 开始日期 'YYYY-MM-DD'
        end_date: str, 结束日期 'YYYY-MM-DD'
        
        Returns:
        --------
        news_df: DataFrame, 新闻数据
        """
        # 新浪财经API（示例，实际需替换为可用接口）
        base_url = 'http://vip.stock.finance.sina.com.cn/corp/go.php/vCB_AllNewsStock/symbol/'
        
        news_list = []
        current_date = datetime.strptime(start_date, '%Y-%m-%d')
        end_date = datetime.strptime(end_date, '%Y-%m-%d')
        
        while current_date <= end_date:
            try:
                # 构造URL（示例格式，需根据实际API调整）
                url = f"{base_url}{stock_code}/page/1.shtml"
                
                # 发送请求
                response = requests.get(url, headers=self.headers, timeout=10)
                
                if response.status_code == 200:
                    # 解析HTML（使用BeautifulSoup或正则表达式）
                    # 这里简化为示例结构
                    news_item = {
                        'date': current_date.strftime('%Y-%m-%d'),
                        'title': '示例新闻标题',
                        'content': '新闻正文内容...',
                        'source': '新浪财经',
                        'url': url
                    }
                    news_list.append(news_item)
                
                # 移至下一天
                current_date += timedelta(days=1)
                
            except Exception as e:
                print(f"Error collecting news for {current_date}: {e}")
                current_date += timedelta(days=1)
        
        news_df = pd.DataFrame(news_list)
        return news_df
    
    def collect_eastmoney(self, stock_code, start_date, end_date):
        """
        采集东方财富新闻（示例框架）
        """
        # 类似实现...
        pass

# 使用示例
# collector = NewsCollector()
# news_df = collector.collect_sina_finance('600000', '2026-01-01', '2026-06-14')

import jieba
from collections import Counter

class SentimentAnalyzer:
    """情感分析器"""
    
    def __init__(self):
        # 加载情感词典（示例，实际需下载完整词典）
        self.positive_words = set(['上涨', '增长', '盈利', '突破', '利好', '乐观'])
        self.negative_words = set(['下跌', '亏损', '下滑', '风险', '利空', '悲观'])
        
        # 否定词（用于调整情感极性）
        self.negation_words = set(['不', '没', '无', '非', '未'])
        
        # 程度副词（用于调整情感强度）
        self.intensity_words = {
            '非常': 2.0,
            '很': 1.5,
            '较': 1.2,
            '稍微': 0.8
        }
    
    def analyze_dict_based(self, text):
        """
        基于词典的情感分析
        
        Parameters:
        -----------
        text: str, 待分析文本
        
        Returns:
        --------
        sentiment_score: float, 情感得分（-1到1之间）
        """
        # 分词
        words = jieba.lcut(text)
        
        positive_count = 0
        negative_count = 0
        
        negation_flag = False
        intensity_multiplier = 1.0
        
        for i, word in enumerate(words):
            # 检查否定词
            if word in self.negation_words:
                negation_flag = True
                continue
            
            # 检查程度副词
            if word in self.intensity_words:
                intensity_multiplier = self.intensity_words[word]
                continue
            
            # 统计情感词
            if word in self.positive_words:
                score = 1.0 * intensity_multiplier
                if negation_flag:
                    negative_count += score
                else:
                    positive_count += score
                negation_flag = False
                intensity_multiplier = 1.0
            
            elif word in self.negative_words:
                score = 1.0 * intensity_multiplier
                if negation_flag:
                    positive_count += score
                else:
                    negative_count += score
                negation_flag = False
                intensity_multiplier = 1.0
        
        # 计算情感得分
        total_words = len(words)
        if total_words == 0:
            return 0.0
        
        sentiment_score = (positive_count - negative_count) / total_words
        
        # 归一化到[-1, 1]
        sentiment_score = max(-1, min(1, sentiment_score * 10))
        
        return sentiment_score
    
    def analyze_advanced(self, text, model='finbert'):
        """
        基于预训练模型的情感分析
        
        Parameters:
        -----------
        text: str, 待分析文本
        model: str, 模型选择（'finbert' / 'chinese-bert'）
        
        Returns:
        --------
        sentiment_score: float, 情感得分
        """
        if model == 'finbert':
            # 使用FinBERT（金融领域预训练模型）
            # 需要安装: pip install transformers torch
            from transformers import AutoTokenizer, AutoModelForSequenceClassification
            import torch
            
            # 加载模型和分词器
            tokenizer = AutoTokenizer.from_pretrained('yiyanghkust/finbert-tone-chinese')
            model = AutoModelForSequenceClassification.from_pretrained('yiyanghkust/finbert-tone-chinese')
            
            # 编码文本
            inputs = tokenizer(text, return_tensors='pt', truncation=True, max_length=512)
            
            # 预测
            with torch.no_grad():
                outputs = model(**inputs)
                probabilities = torch.softmax(outputs.logits, dim=1)
            
            # 返回正面情感概率 - 负面情感概率
            sentiment_score = probabilities[0][2].item() - probabilities[0][0].item()
            
            return sentiment_score
        
        elif model == 'chinese-bert':
            # 使用中文BERT（类似实现）
            pass
        
        else:
            raise ValueError(f"Unsupported model: {model}")

# 使用示例
# analyzer = SentimentAnalyzer()
# score = analyzer.analyze_dict_based("公司业绩大幅增长，盈利能力显著提升")
# print(f"Sentiment Score: {score}")  # 应输出正值

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report

class MLSentimentAnalyzer:
    """基于机器学习的情感分析器"""
    
    def __init__(self):
        self.vectorizer = TfidfVectorizer(max_features=5000, ngram_range=(1, 2))
        self.classifier = LogisticRegression(random_state=42)
    
    def prepare_training_data(self, news_df, label_column='sentiment'):
        """
        准备训练数据
        
        Parameters:
        -----------
        news_df: DataFrame, 包含新闻文本和标签的数据框
        label_column: str, 标签列名（'positive' / 'negative' / 'neutral'）
        
        Returns:
        --------
        X: array, 特征矩阵
        y: array, 标签向量
        """
        # 文本向量化
        X = self.vectorizer.fit_transform(news_df['content'])
        
        # 标签编码
        y = news_df[label_column].map({
            'positive': 2,
            'neutral': 1,
            'negative': 0
        })
        
        return X, y
    
    def train(self, X, y):
        """
        训练分类器
        """
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42
        )
        
        self.classifier.fit(X_train, y_train)
        
        # 评估
        y_pred = self.classifier.predict(X_test)
        print(classification_report(y_test, y_pred))
    
    def predict(self, texts):
        """
        预测新文本的情感
        
        Parameters:
        -----------
        texts: list, 文本列表
        
        Returns:
        --------
        probabilities: array, 情感概率（negative, neutral, positive）
        """
        X = self.vectorizer.transform(texts)
        probabilities = self.classifier.predict_proba(X)
        
        return probabilities

# 使用示例
# ml_analyzer = MLSentimentAnalyzer()
# X, y = ml_analyzer.prepare_training_data(labeled_news_df)
# ml_analyzer.train(X, y)
# probs = ml_analyzer.predict(["公司业绩大幅增长"])

import pandas as pd
import numpy as np

class EventDrivenStrategy:
    """事件驱动策略"""
    
    def __init__(self, sentiment_analyzer, lookback=5, holding_period=10):
        """
        初始化策略
        
        Parameters:
        -----------
        sentiment_analyzer: object, 情感分析器
        lookback: int, 情感得分回溯期（天）
        holding_period: int, 持仓期限（天）
        """
        self.analyzer = sentiment_analyzer
        self.lookback = lookback
        self.holding_period = holding_period
        
    def calculate_sentiment_signal(self, news_df, method='dict'):
        """
        计算情感信号
        
        Parameters:
        -----------
        news_df: DataFrame, 新闻数据（包含'date', 'title', 'content'列）
        method: str, 情感分析方法（'dict' / 'ml' / 'finbert'）
        
        Returns:
        --------
        signal_df: DataFrame, 情感信号（日期 x 股票）
        """
        # 按日期和股票分组
        grouped = news_df.groupby(['date', 'stock_code'])
        
        signal_list = []
        
        for (date, stock), group in grouped:
            # 汇总该股票在该日期的所有新闻
            all_text = ' '.join(group['title'].tolist() + group['content'].tolist())
            
            # 情感分析
            if method == 'dict':
                sentiment_score = self.analyzer.analyze_dict_based(all_text)
            elif method == 'finbert':
                sentiment_score = self.analyzer.analyze_advanced(all_text, model='finbert')
            else:
                raise ValueError(f"Unsupported method: {method}")
            
            signal_list.append({
                'date': date,
                'stock_code': stock,
                'sentiment_score': sentiment_score,
                'news_count': len(group)
            })
        
        signal_df = pd.DataFrame(signal_list)
        
        # 透视表：行为日期，列为股票，值为情感得分
        signal_df = signal_df.pivot(index='date', columns='stock_code', values='sentiment_score')
        
        return signal_df
    
    def generate_trading_signal(self, sentiment_signals, prices):
        """
        生成交易信号
        
        Parameters:
        -----------
        sentiment_signals: DataFrame, 情感信号
        prices: DataFrame, 股票价格数据
        
        Returns:
        --------
        positions: DataFrame, 持仓信号（1=多仓, -1=空仓, 0=平仓）
        """
        # 标准化情感得分（横截面Z-score）
        standardized_signals = sentiment_signals.apply(
            lambda x: (x - x.mean()) / x.std(), axis=1
        )
        
        # 生成信号：情感得分最高的前10%做多，最低的前10%做空
        positions = pd.DataFrame(index=standardized_signals.index, 
                                columns=standardized_signals.columns, 
                                data=0)
        
        for date in standardized_signals.index:
            signals = standardized_signals.loc[date].dropna()
            
            if len(signals) == 0:
                continue
            
            # 多仓：情感得分 > 90分位数
            long_threshold = np.percentile(signals, 90)
            long_stocks = signals[signals > long_threshold].index
            
            # 空仓：情感得分 < 10分位数
            short_threshold = np.percentile(signals, 10)
            short_stocks = signals[signals < short_threshold].index
            
            positions.loc[date, long_stocks] = 1
            positions.loc[date, short_stocks] = -1
        
        return positions
    
    def backtest(self, positions, prices, transaction_cost=0.003):
        """
        回测策略
        
        Parameters:
        -----------
        positions: DataFrame, 持仓信号
        prices: DataFrame, 股票价格数据
        transaction_cost: float, 交易成本（双边）
        
        Returns:
        --------
        performance: dict, 绩效指标
        """
        # 计算收益率
        returns = prices.pct_change()
        
        # 策略收益率
        strategy_returns = (positions.shift(1) * returns).sum(axis=1) / (positions.shift(1) != 0).sum(axis=1)
        
        # 扣除交易成本
        turnover = positions.diff().abs().sum(axis=1) / 2
        strategy_returns -= turnover * transaction_cost
        
        # 累计收益
        cumulative_returns = (1 + strategy_returns).cumprod()
        
        # 绩效指标
        total_return = cumulative_returns.iloc[-1] - 1
        annual_return = (1 + total_return) ** (252 / len(strategy_returns)) - 1
        annual_vol = strategy_returns.std() * np.sqrt(252)
        sharpe = annual_return / annual_vol if annual_vol != 0 else 0
        max_drawdown = ((cumulative_returns / cumulative_returns.cummax()) - 1).min()
        
        performance = {
            'total_return': total_return,
            'annual_return': annual_return,
            'annual_volatility': annual_vol,
            'sharpe_ratio': sharpe,
            'max_drawdown': max_drawdown,
            'strategy_returns': strategy_returns,
            'cumulative_returns': cumulative_returns
        }
        
        return performance

# 使用示例
# strategy = EventDrivenStrategy(sentiment_analyzer)
# sentiment_signals = strategy.calculate_sentiment_signal(news_df)
# positions = strategy.generate_trading_signal(sentiment_signals, prices)
# performance = strategy.backtest(positions, prices)

# 完整回测代码（简化版）

# 1. 加载数据
news_df = pd.read_csv('data/news_data_2018_2026.csv')
prices = pd.read_csv('data/stock_prices_2018_2026.csv', index_col='date')

# 2. 情感分析
analyzer = SentimentAnalyzer()
strategy = EventDrivenStrategy(analyzer, lookback=5, holding_period=5)

sentiment_signals = strategy.calculate_sentiment_signal(news_df, method='finbert')

# 3. 生成交易信号
positions = strategy.generate_trading_signal(sentiment_signals, prices)

# 4. 回测
performance = strategy.backtest(positions, prices, transaction_cost=0.003)

# 5. 结果展示
print("=== 新闻情感策略回测结果 ===")
print(f"年化收益率: {performance['annual_return']*100:.2f}%")
print(f"年化波动率: {performance['annual_volatility']*100:.2f}%")
print(f"夏普比率: {performance['sharpe_ratio']:.2f}")
print(f"最大回撤: {performance['max_drawdown']*100:.2f}%")

# 6. 可视化
import matplotlib.pyplot as plt

fig, axes = plt.subplots(2, 1, figsize=(15, 10))

# 累计收益曲线
cumulative_returns = performance['cumulative_returns']
axes[0].plot(cumulative_returns.index, cumulative_returns.values)
axes[0].set_title('News Sentiment Strategy - Cumulative Returns')
axes[0].set_xlabel('Date')
axes[0].set_ylabel('Cumulative Return')

# 回撤曲线
drawdown = (cumulative_returns / cumulative_returns.cummax()) - 1
axes[1].fill_between(drawdown.index, 0, drawdown.values, alpha=0.3, color='red')
axes[1].plot(drawdown.index, drawdown.values, color='red')
axes[1].set_title('Drawdown')
axes[1].set_xlabel('Date')
axes[1].set_ylabel('Drawdown')

plt.tight_layout()
plt.savefig('news_sentiment_backtest.png', dpi=300, bbox_inches='tight')

class MultiDimensionalSentimentAnalyzer:
    """多维度情感分析器"""
    
    def __init__(self):
        # 加载多维度词典
        self.dimension_dicts = {
            'sentiment': {...},  # 传统情感词典
            'certainty': {'确认': 1, '肯定': 1, '可能': 0.5, '或许': 0.3},
            'timeliness': {'紧急': 1, '立即': 1, '近期': 0.7, '未来': 0.3},
            'scope': {'行业': 0.8, '市场': 1, '公司': 0.5, '全球': 1.2}
        }
    
    def analyze_multi_dimension(self, text):
        """
        多维度情感分析
        
        Returns:
        --------
        scores: dict, 各维度的得分
        """
        words = jieba.lcut(text)
        
        scores = {}
        for dimension, word_dict in self.dimension_dicts.items():
            score = self._calculate_dimension_score(words, word_dict)
            scores[dimension] = score
        
        return scores

def enhance_sentiment_signal(sentiment_signals, method='momentum'):
    """
    增强情感信号
    
    Parameters:
    -----------
    sentiment_signals: DataFrame, 原始情感信号
    method: str, 增强方法
    
    Returns:
    --------
    enhanced_signals: DataFrame, 增强后的信号
    """
    if method == 'momentum':
        # 情感动量：当前情感 - 过去N天平均情感
        enhanced_signals = sentiment_signals - sentiment_signals.rolling(5).mean()
    
    elif method == 'volatility':
        # 情感波动率：过去N天情感的标准差
        enhanced_signals = sentiment_signals.rolling(5).std()
    
    elif method == 'consensus':
        # 情感共识：多源新闻情感的离散度（越低越一致）
        # 需要多源数据
        pass
    
    return enhanced_signals

def filter_by_technical_indicators(positions, prices, volume):
    """
    使用技术指标过滤交易信号
    
    Parameters:
    -----------
    positions: DataFrame, 原始持仓信号
    prices: DataFrame, 价格数据
    volume: DataFrame, 成交量数据
    
    Returns:
    --------
    filtered_positions: DataFrame, 过滤后的持仓信号
    """
    filtered_positions = positions.copy()
    
    for stock in positions.columns:
        # 计算技术指标
        ma20 = prices[stock].rolling(20).mean()
        ma60 = prices[stock].rolling(60).mean()
        volume_ma20 = volume[stock].rolling(20).mean()
        
        for date in positions.index:
            # 过滤条件1：股价在20日均线下方时不做多
            if positions.loc[date, stock] == 1 and prices[stock].loc[date] < ma20.loc[date]:
                filtered_positions.loc[date, stock] = 0
            
            # 过滤条件2：成交量未放大时不做多（缺乏确认）
            if positions.loc[date, stock] == 1 and volume[stock].loc[date] < volume_ma20.loc[date]:
                filtered_positions.loc[date, stock] = 0
    
    return filtered_positions

def dynamic_holding_period(positions, sentiment_signals, min_hold=3, max_hold=15):
    """
    动态持仓期限
    
    Parameters:
    -----------
    positions: DataFrame, 持仓信号
    sentiment_signals: DataFrame, 情感信号强度
    min_hold: int, 最小持仓天数
    max_hold: int, 最大持仓天数
    
    Returns:
    --------
    adjusted_positions: DataFrame, 调整后的持仓信号
    """
    adjusted_positions = positions.copy()
    
    for stock in positions.columns:
        entry_date = None
        hold_days = 0
        
        for date in positions.index:
            if positions.loc[date, stock] != 0:
                if entry_date is None:
                    entry_date = date
                    hold_days = 0
                
                # 根据情感强度决定持有天数
                sentiment_intensity = abs(sentiment_signals.loc[date, stock])
                target_hold = int(min_hold + (max_hold - min_hold) * sentiment_intensity)
                
                hold_days += 1
                
                # 达到目标持有天数，平仓
                if hold_days >= target_hold:
                    adjusted_positions.loc[date, stock] = 0
                    entry_date = None
        
    return adjusted_positions

class NewsDataQualityControl:
    """新闻数据质量控制"""
    
    def __init__(self):
        self.min_title_length = 10  # 标题最小长度
        self.max_title_length = 200  # 标题最大长度
        self.duplicate_threshold = 0.8  # 相似度阈值
    
    def clean_news_data(self, news_df):
        """
        清洗新闻数据
        
        Parameters:
        -----------
        news_df: DataFrame, 原始新闻数据
        
        Returns:
        --------
        cleaned_df: DataFrame, 清洗后的数据
        """
        # 1. 去除标题过短或过长的新闻
        cleaned_df = news_df[
            (news_df['title'].str.len() >= self.min_title_length) &
            (news_df['title'].str.len() <= self.max_title_length)
        ]
        
        # 2. 去除重复新闻（基于标题相似度）
        cleaned_df = self._remove_duplicates(cleaned_df)
        
        # 3. 区分新闻与公告
        cleaned_df = self._classify_news_type(cleaned_df)
        
        # 4. 去除垃圾信息（广告、爬虫错误等）
        cleaned_df = self._remove_spam(cleaned_df)
        
        return cleaned_df
    
    def _remove_duplicates(self, news_df):
        """去除重复新闻"""
        from difflib import SequenceMatcher
        
        unique_news = []
        titles = news_df['title'].tolist()
        
        for i, row in news_df.iterrows():
            is_duplicate = False
            
            for unique_title in [n['title'] for n in unique_news]:
                similarity = SequenceMatcher(None, row['title'], unique_title).ratio()
                
                if similarity > self.duplicate_threshold:
                    is_duplicate = True
                    break
            
            if not is_duplicate:
                unique_news.append(row.to_dict())
        
        return pd.DataFrame(unique_news)
    
    def _classify_news_type(self, news_df):
        """区分新闻与公告"""
        # 基于关键词分类
        announcement_keywords = ['公告', '报告书', '决议', '通知']
        
        news_df['type'] = news_df['title'].apply(
            lambda x: 'announcement' if any(kw in x for kw in announcement_keywords) else 'news'
        )
        
        return news_df
    
    def _remove_spam(self, news_df):
        """去除垃圾信息"""
        # 广告关键词
        spam_keywords = ['加微信', 'QQ群', '荐股', '牛股']
        
        spam_mask = news_df['title'].apply(
            lambda x: any(kw in x for kw in spam_keywords)
        )
        
        return news_df[~spam_mask]

class EnsembleSentimentAnalyzer:
    """集成情感分析器"""
    
    def __init__(self):
        self.analyzers = {
            'dict': SentimentAnalyzer(),
            'ml': MLSentimentAnalyzer(),
            'finbert': None  # 延迟加载
        }
        self.weights = {'dict': 0.3, 'ml': 0.3, 'finbert': 0.4}
    
    def analyze(self, text):
        """
        集成情感分析
        
        Returns:
        --------
        ensemble_score: float, 集成情感得分
        """
        scores = {}
        
        # 词典法
        scores['dict'] = self.analyzers['dict'].analyze_dict_based(text)
        
        # 机器学习法
        # scores['ml'] = self.analyzers['ml'].predict([text])[0]
        
        # FinBERT
        if self.analyzers['finbert'] is None:
            # 延迟加载
            from transformers import AutoTokenizer, AutoModelForSequenceClassification
            self.analyzers['finbert'] = {
                'tokenizer': AutoTokenizer.from_pretrained('yiyanghkust/finbert-tone-chinese'),
                'model': AutoModelForSequenceClassification.from_pretrained('yiyanghkust/finbert-tone-chinese')
            }
        
        scores['finbert'] = self._analyze_finbert(text)
        
        # 加权平均
        ensemble_score = sum(
            scores[method] * self.weights[method] 
            for method in scores
        )
        
        return ensemble_score
    
    def _analyze_finbert(self, text):
        """使用FinBERT分析"""
        tokenizer = self.analyzers['finbert']['tokenizer']
        model = self.analyzers['finbert']['model']
        
        inputs = tokenizer(text, return_tensors='pt', truncation=True, max_length=512)
        
        with torch.no_grad():
            outputs = model(**inputs)
            probabilities = torch.softmax(outputs.logits, dim=1)
        
        sentiment_score = probabilities[0][2].item() - probabilities[0][0].item()
        
        return sentiment_score

def reduce_turnover(positions, min_holding_days=5):
    """
    降低调仓频率
    
    Parameters:
    -----------
    positions: DataFrame, 原始持仓信号
    min_holding_days: int, 最小持有天数
    
    Returns:
    --------
    adjusted_positions: DataFrame, 调整后的持仓信号
    """
    adjusted_positions = positions.copy()
    
    for stock in positions.columns:
        last_change_date = None
        
        for i, date in enumerate(positions.index):
            if i == 0:
                continue
            
            # 检测持仓变化
            if positions.loc[date, stock] != positions.iloc[i-1][stock]:
                if last_change_date is None:
                    last_change_date = date
                else:
                    # 计算距离上次调仓的天数
                    days_since_last_change = (date - last_change_date).days
                    
                    if days_since_last_change < min_holding_days:
                        # 维持原有持仓
                        adjusted_positions.loc[date, stock] = adjusted_positions.iloc[i-1][stock]
                    else:
                        last_change_date = date
        
    return adjusted_positions