另类数据：量化投资的新阿尔法源泉——从卫星图像到社交情绪

import tweepy
import pandas as pd
from transformers import pipeline

# 初始化Twitter API (v2)
client = tweepy.Client(bearer_token='YOUR_BEARER_TOKEN')

# 金融情感分析模型 (FinBERT)
sentiment_analyzer = pipeline(
    "sentiment-analysis",
    model="yiyanghkust/finbert-tone",
    tokenizer="yiyanghkust/finbert-tone"
)

def get_stock_sentiment(stock_ticker, max_results=100):
    """
    获取股票在Twitter上的情绪得分
    """
    # 搜索推文
    query = f"${stock_ticker} lang:en -is:retweet"
    tweets = client.search_recent_tweets(
        query=query,
        max_results=max_results,
        tweet_fields=['created_at']
    )
    
    if not tweets.data:
        return None
    
    # 情感分析
    sentiments = []
    for tweet in tweets.data:
        result = sentiment_analyzer(tweet.text)[0]
        # Positive: +1, Negative: -1, Neutral: 0
        score = {'Positive': 1, 'Negative': -1, 'Neutral': 0}[result['label']]
        sentiments.append(score)
    
    # 计算平均情绪得分 (-1 到 1)
    avg_sentiment = np.mean(sentiments)
    return avg_sentiment

# 使用示例
aapl_sentiment = get_stock_sentiment('AAPL')
print(f"AAPL Twitter情绪得分: {aapl_sentiment:.3f}")

# 示例：卫星图像因子
import numpy as np
import pandas as pd

def build_satellite_factor(stock_universe, date_range):
    """
    构建基于卫星图像的另类数据因子
    
    输入：
    - stock_universe: 股票池（如沪深300成分股）
    - date_range: 时间范围
    
    输出：
    - factor_df: 因子值 DataFrame (date x stock)
    """
    factor_df = pd.DataFrame(index=date_range, columns=stock_universe)
    
    for date in date_range:
        for stock in stock_universe:
            # 获取该股票对应公司的停车场饱和度
            parking_score = get_parking_capacity(stock, date)  # 自定义函数
            
            # 获取该股票对应公司的原油库存变化（如果是能源股）
            if is_energy_sector(stock):
                inventory_change = get_oil_inventory_change(stock, date)
                factor_df.loc[date, stock] = 0.6 * parking_score + 0.4 * inventory_change
            else:
                factor_df.loc[date, stock] = parking_score
    
    # 横截面标准化
    factor_df = factor_df.rank(axis=1, pct=True)
    
    return factor_df

# 将另类数据因子整合到多因子模型
def multi_factor_model(fundamental_factor, satellite_factor, momentum_factor, weights=[0.4, 0.3, 0.3]):
    """
    多因子模型：基本面 + 另类数据 + 动量
    """
    combined_factor = (weights[0] * fundamental_factor + 
                      weights[1] * satellite_factor + 
                      weights[2] * momentum_factor)
    return combined_factor

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

tokenizer = AutoTokenizer.from_pretrained("yiyanghkust/finbert-tone")
model = AutoModelForSequenceClassification.from_pretrained("yiyanghkust/finbert-tone")

def extract_financial_sentiment(text):
    """
    使用 FinBERT 提取金融文本情感
    """
    inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
    outputs = model(**inputs)
    probabilities = torch.nn.functional.softmax(outputs.logits, dim=-1)
    
    # 输出：Positive, Negative, Neutral 的概率
    sentiment = ['Positive', 'Negative', 'Neutral']
    scores = {sentiment[i]: probabilities[0][i].item() for i in range(3)}
    
    return scores

# 示例：分析财经新闻
news = "Apple announces record quarterly revenue, beating analyst expectations."
sentiment_scores = extract_financial_sentiment(news)
print(sentiment_scores)
# {'Positive': 0.89, 'Negative': 0.02, 'Neutral': 0.09}

import torch
from torchvision import models, transforms
from PIL import Image

def count_cars_in_parking_lot(satellite_image_path):
    """
    使用目标检测模型（如YOLO）统计停车场车辆数
    """
    # 加载预训练的YOLO模型
    model = torch.hub.load('ultralytics/yolov5', 'yolov5s', pretrained=True)
    
    # 推理
    results = model(satellite_image_path)
    
    # 统计车辆数（COCO数据集中car的类别ID=2）
    car_count = len([r for r in results.xyxy[0] if int(r[-1]) == 2])
    
    return car_count

# 示例：分析沃尔玛停车场
walmart_parking = count_cars_in_parking_lot('walmart_parking_lot_satellite.jpg')
print(f"停车场车辆数: {walmart_parking}")

class EventDrivenStrategy:
    """
    基于另类数据的事件驱动策略
    """
    def __init__(self, news_api, sentiment_threshold=0.8):
        self.news_api = news_api
        self.threshold = sentiment_threshold
        
    def on_news_received(self, news_item):
        """
        接收到新闻后的交易逻辑
        """
        # 1. 提取情感得分
        sentiment = extract_financial_sentiment(news_item['content'])
        
        # 2. 判断是否为极端情感事件
        if sentiment['Positive'] > self.threshold:
            # 强烈正面新闻 → 做多
            self.execute_trade(news_item['stock'], 'BUY', position_size=0.05)
        elif sentiment['Negative'] > self.threshold:
            # 强烈负面新闻 → 做空
            self.execute_trade(news_item['stock'], 'SELL', position_size=0.05)
        
        # 3. 设置止损/止盈
        self.set_risk_management(news_item['stock'])
    
    def execute_trade(self, stock, direction, position_size):
        """
        执行交易（简化示例）
        """
        print(f"事件驱动交易: {direction} {stock} {position_size*100}% 仓位")