另类数据：量化交易的下一个阿尔法源泉

┌─────────────────────────────────────────┐
│  1. 卫星/地理数据（Geolocation Data）    │
│     - 卫星图像                           │
│     - 手机定位数据                       │
│     - 停车场车辆计数                     │
└─────────────────────────────────────────┘
           ↓
┌─────────────────────────────────────────┐
│  2. 社交媒体/舆情数据（Sentiment Data）  │
│     - Twitter/StockTwits情绪            │
│     - 新闻情感分析                       │
│     - Reddit/雪球讨论热度               │
└─────────────────────────────────────────┘
           ↓
┌─────────────────────────────────────────┐
│  3. 消费/交易数据（Consumer Data）      │
│     - 信用卡交易数据                     │
│     - 电商销售数据                       │
│     - 应用程序使用数据                   │
└─────────────────────────────────────────┘

import numpy as np
import pandas as pd
from PIL import Image
import requests
from io import BytesIO

def count_parking_cars(satellite_image_url):
    """
    用卫星图像计算停车场车辆数
    
    实际中会用计算机视觉（YOLO/ResNet）做目标检测
    这里简化为示例
    """
    # 下载卫星图像
    response = requests.get(satellite_image_url)
    img = Image.open(BytesIO(response.content))
    
    # 计算机视觉模型（简化）
    # model = YOLO('car_detector.pt')
    # results = model(img)
    # car_count = len(results.xyxy[0])
    
    # 示例：返回模拟数据
    car_count = np.random.randint(50, 200)
    
    return car_count

# 示例：追踪沃尔玛停车场的车辆数
walmart_stores = ['store_001', 'store_002', 'store_003']

parking_counts = {}
for store in walmart_stores:
    # 每周拍摄一次
    img_url = f"https://satellite-api.com/walmart/{store}/latest.jpg"
    count = count_parking_cars(img_url)
    parking_counts[store] = count

# 汇总所有门店
total_cars = sum(parking_counts.values())

# 与历史销售额回归
# sales = alpha + beta * total_cars + error

def estimate_oil_inventory(satellite_images):
    """
    通过卫星图像估算原油库存
    
    关键技术：测量储油罐的阴影长度 → 计算浮顶位置 → 推算油量
    """
    inventory_estimates = []
    
    for img in satellite_images:
        # 步骤1：检测储油罐
        tanks = detect_storage_tanks(img)
        
        # 步骤2：计算每个储油罐的油位
        for tank in tanks:
            shadow_length = measure_shadow(tank)
            oil_level = calculate_oil_level(shadow_length)
            capacity = tank['capacity']
            oil_volume = oil_level * capacity
            
            inventory_estimates.append(oil_volume)
    
    total_inventory = sum(inventory_estimates)
    
    return total_inventory

# 实际应用
# EIA（美国能源信息署）每周三公布库存数据
# 卫星数据可以提前3-5天预测，获得交易优势

def analyze_foot_traffic(stores, date_range):
    """
    分析门店客流量
    
    数据来源：手机定位数据提供商（如SafeGraph, Cuebiq）
    """
    foot_traffic = {}
    
    for store in stores:
        # 获取该门店周边的手机定位数据
        location_data = get_mobile_locations(store['latitude'], 
                                            store['longitude'], 
                                            radius=500)  # 500米半径
        
        # 去重（同一设备的多次记录）
        unique_devices = location_data['device_id'].nunique()
        
        foot_traffic[store['name']] = unique_devices
    
    return foot_traffic

# 示例：对比星巴克 vs 瑞幸咖啡的客流量
starbucks_traffic = analyze_foot_traffic(starbucks_stores, '2024-01-01', '2024-01-31')
luckin_traffic = analyze_foot_traffic(luckin_stores, '2024-01-01', '2024-01-31')

# 如果瑞幸客流量增长快于星巴克 → 做多瑞幸，做空星巴克

import tweepy
from textblob import TextBlob
import pandas as pd

def twitter_sentiment_analysis(stock_symbol, n_tweets=1000):
    """
    Twitter情绪分析
    
    实际中会用更先进的模型（BERT, FinBERT）
    """
    # 认证Twitter API
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)
    api = tweepy.API(auth)
    
    # 搜索相关推文
    tweets = tweepy.Cursor(api.search_tweets, 
                           q=f"${stock_symbol} -filter:retweets", 
                           lang="en").items(n_tweets)
    
    sentiments = []
    for tweet in tweets:
        # 情感分析
        analysis = TextBlob(tweet.text)
        sentiment_score = analysis.sentiment.polarity
        sentiments.append(sentiment_score)
    
    # 汇总情绪指数
    avg_sentiment = np.mean(sentiments)
    sentiment_std = np.std(sentiments)
    
    return {
        'avg_sentiment': avg_sentiment,
        'std': sentiment_std,
        'n_tweets': len(sentiments)
    }

# 示例：分析特斯拉的Twitter情绪
tsla_sentiment = twitter_sentiment_analysis('TSLA', n_tweets=5000)

print(f"特斯拉Twitter情绪指数: {tsla_sentiment['avg_sentiment']:.3f}")
print(f"样本量: {tsla_sentiment['n_tweets']}")

# 交易信号
# if sentiment < -0.5: 极度悲观 → 买入（逆向策略）
# if sentiment > 0.5: 极度乐观 → 卖出

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

def finbert_sentiment(texts):
    """
    用FinBERT做金融情感分析
    
    FinBERT在金融文本上预训练，比通用NLP模型更准确
    """
    tokenizer = AutoTokenizer.from_pretrained("yiyanghkust/finbert-tone")
    model = AutoModelForSequenceClassification.from_pretrained("yiyanghkust/finbert-tone")
    
    # 批量处理
    inputs = tokenizer(texts, padding=True, truncation=True, return_tensors="pt")
    
    with torch.no_grad():
        outputs = model(**inputs)
    
    predictions = torch.argmax(outputs.logits, dim=-1)
    
    # 0: 负面, 1: 中性, 2: 正面
    sentiment_map = {0: 'negative', 1: 'neutral', 2: 'positive'}
    results = [sentiment_map[p.item()] for p in predictions]
    
    return results

# 示例
tweets = [
    "Tesla's delivery numbers blew past expectations!",
    "Concerned about Tesla's margin compression",
    "TSLA is fairly valued at current price"
]

sentiments = finbert_sentiment(tweets)
print(sentiments)  # ['positive', 'negative', 'neutral']

import praw  # Reddit API
import pandas as pd
from datetime import datetime, timedelta

def reddit_mention_count(stock_symbol, subreddit='wallstreetbets'):
    """
    统计Reddit上某股票的提及次数
    
    关键洞察：
    - 提及次数暴增 → 散户FOMO → 短期上涨
    - 但散户涌入后往往下跌（聪明钱出货）
    """
    # 连接Reddit API
    reddit = praw.Reddit(client_id='', client_secret='', user_agent='')
    
    subreddit = reddit.subreddit(subreddit)
    
    # 搜索提及
    mentions = []
    for submission in subreddit.new(limit=1000):
        if stock_symbol.lower() in submission.title.lower():
            mentions.append({
                'title': submission.title,
                'score': submission.score,
                'created': datetime.fromtimestamp(submission.created_utc),
                'num_comments': submission.num_comments
            })
    
    df = pd.DataFrame(mentions)
    
    # 计算热度指标
    if not df.empty:
        total_mentions = len(df)
        avg_score = df['score'].mean()
        total_comments = df['num_comments'].sum()
        
        hype_score = total_mentions * avg_score * np.log1p(total_comments)
        
        return {
            'mentions': total_mentions,
            'hype_score': hype_score,
            'recent_mentions': df[df['created'] > datetime.now() - timedelta(days=1)]
        }
    
    return None

# 示例：监控GME（GameStop）的Reddit热度
gme_hype = reddit_mention_count('GME', subreddit='wallstreetbets')

print(f"GME在WSB的提及次数: {gme_hype['mentions']}")
print(f"热度分数: {gme_hype['hype_score']:.0f}")

# 交易信号
# if hype_score > threshold: 散户FOMO严重 → 考虑做空（反转策略）

def news_sentiment_event_driven(stock_symbol, lookback_days=7):
    """
    事件驱动情感分析
    
    策略：重大新闻发布后，情感极端 → 短期价格压力 → 均值回归
    """
    from newsapi import NewsApiClient
    
    newsapi = NewsApiClient(api_key='your_api_key')
    
    # 获取新闻
    all_articles = newsapi.get_everything(q=stock_symbol,
                                          from_param=datetime.now() - timedelta(days=lookback_days),
                                          language='en',
                                          sort_by='publishedAt')
    
    sentiments = []
    for article in all_articles['articles']:
        title = article['title']
        description = article.get('description', '')
        text = title + ' ' + description
        
        # 情感分析
        sentiment = finbert_sentiment([text])[0]
        sentiments.append({
            'sentiment': sentiment,
            'published_at': article['publishedAt'],
            'title': title
        })
    
    # 汇总
    sentiment_counts = pd.DataFrame(sentiments)['sentiment'].value_counts()
    
    # 计算情感分数（-1到1）
    sentiment_score = (sentiment_counts.get('positive', 0) - sentiment_counts.get('negative', 0)) / len(sentiments)
    
    return sentiment_score, sentiments

# 示例：分析苹果公司最近的新闻情感
aapl_sentiment, articles = news_sentiment_event_driven('AAPL', lookback_days=3)

print(f"苹果新闻情感分数: {aapl_sentiment:.2f}")

# if sentiment_score < -0.5: 负面新闻暴增 → 短期超卖 → 买入（均值回归）
# if sentiment_score > 0.5: 正面新闻暴增 → 短期超买 → 卖出

def credit_card_transaction_data(retailers, quarter='2024Q1'):
    """
    信用卡交易数据分析
    
    数据来源：另类数据提供商（如Facteus, Second Measure）
    实际中需要购买数据授权
    """
    # 模拟数据（实际中从数据提供商API获取）
    transaction_data = {
        'WMT': {'transaction_count': 1000000, 'avg_amount': 45.20},  # 沃尔玛
        'TGT': {'transaction_count': 500000, 'avg_amount': 62.50},  # Target
        'COST': {'transaction_count': 200000, 'avg_amount': 150.30}  # Costco
    }
    
    sales_forecast = {}
    for retailer in retailers:
        if retailer in transaction_data:
            data = transaction_data[retailer]
            # 预测销售额 = 交易笔数 × 平均金额
            forecasted_sales = data['transaction_count'] * data['avg_amount']
            sales_forecast[retailer] = forecasted_sales
    
    return sales_forecast

# 示例：预测沃尔玛Q1销售额
wmt_forecast = credit_card_transaction_data(['WMT'], quarter='2024Q1')

# 与实际财报对比
# if forecasted > analyst_consensus: 做多
# if forecasted < analyst_consensus: 做空

import requests
from bs4 import BeautifulSoup
import pandas as pd

def scrape_amazon_sales_rank(product_asin):
    """
    抓取Amazon销售排名
    
    逻辑：
    - 销售排名上升 → 销量增加 → 收入增加
    - 可用于预测上市公司的电商渠道收入
    """
    url = f"https://www.amazon.com/dp/{product_asin}"
    
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    
    response = requests.get(url, headers=headers)
    soup = BeautifulSoup(response.content, 'html.parser')
    
    # 提取销售排名
    sales_rank_elem = soup.find('li', {'id': 'SalesRank'})
    if sales_rank_elem:
        sales_rank = sales_rank_elem.text.strip()
        # 解析排名数字
        rank_number = int(''.join(filter(str.isdigit, sales_rank)))
        return rank_number
    
    return None

# 示例：监控Nike在Amazon的销售排名
nike_asins = ['B08N5WRWNW', 'B09R4NMJ4H', 'B08N5WRQYX']  # Nike产品ASIN

sales_ranks = {}
for asin in nike_asins:
    rank = scrape_amazon_sales_rank(asin)
    sales_ranks[asin] = rank
    time.sleep(1)  # 避免被封

# 计算平均排名（排名越小 = 销量越高）
avg_rank = np.mean(list(sales_ranks.values()))

# 如果平均排名上升（数字变小）→ Nike销量增加 → 做多Nike

def app_download_tracking(app_ids, date_range):
    """
    追踪APP下载量
    
    可用于预测移动互联网公司的收入
    """
    # 实际中需要用App Store/Google Play的API
    # 或使用第三方数据提供商（如App Annie, Sensor Tower）
    
    download_data = {}
    for app_id in app_ids:
        # 模拟数据
        downloads = np.random.randint(10000, 100000)
        active_users = np.random.randint(100000, 1000000)
        
        download_data[app_id] = {
            'daily_downloads': downloads,
            'mau': active_users  # 月活跃用户
        }
    
    return download_data

# 示例：预测腾讯手游收入（通过APP数据）
tencent_games = ['PUBG_Mobile', 'Honor_of_Kings']

game_data = app_download_tracking(tencent_games, '2024-01-01')

# 如果下载量和MAU增长 → 预测手游收入增长 → 做多腾讯

def clean_sentiment_data(tweets):
    """
    清洗社交媒体数据
    
    步骤：
    1. 去除机器人账号
    2. 去除重复内容
    3. 识别讽刺/反语（用NLP模型）
    """
    cleaned = []
    
    for tweet in tweets:
        # 1. 检测机器人（发帖频率、内容重复度）
        if is_bot(tweet['user_id']):
            continue
        
        # 2. 去重
        if is_duplicate(tweet['text'], cleaned):
            continue
        
        # 3. 讽刺检测
        if detect_sarcasm(tweet['text']):
            # 反转情感标签
            tweet['sentiment'] = reverse_sentiment(tweet['sentiment'])
        
        cleaned.append(tweet)
    
    return cleaned

def is_bot(user_id):
    """
    检测机器人账号
    """
    user_data = get_user_metadata(user_id)
    
    # 特征：发帖频率、关注数/粉丝数比例、账号年龄
    if user_data['posts_per_day'] > 50:
        return True
    if user_data['followers'] / max(user_data['following'], 1) < 0.01:
        return True
    if user_data['account_age_days'] < 30:
        return True
    
    return False

def test_alternative_data_signal(alternative_data, returns, min_samples=1000):
    """
    严格测试另类数据信号
    
    检验清单：
    1. 样本外测试（Out-of-Sample）
    2. 经济逻辑合理性
    3. 过拟合检验（如White Realized Ratio）
    """
    from sklearn.model_selection import TimeSeriesSplit
    from sklearn.metrics import information_coefficient
    
    # 1. 时间序列交叉验证
    tscv = TimeSeriesSplit(n_splits=5)
    ic_scores = []
    
    for train_idx, test_idx in tscv.split(alternative_data):
        X_train, X_test = alternative_data.iloc[train_idx], alternative_data.iloc[test_idx]
        y_train, y_test = returns.iloc[train_idx], returns.iloc[test_idx]
        
        # 训练模型
        model = train_model(X_train, y_train)
        
        # 样本外预测
        y_pred = model.predict(X_test)
        
        # 信息系数（IC）
        ic = information_coefficient(y_test, y_pred)
        ic_scores.append(ic)
    
    avg_ic = np.mean(ic_scores)
    ic_std = np.std(ic_scores)
    
    # 2. 经济逻辑检查
    economic_sense = check_economic_logic(alternative_data, returns)
    
    # 3. 过拟合检验
    from arch.bootstrap import StationaryBootstrap
    # 计算White Realized Ratio
    # ...
    
    return {
        'avg_ic': avg_ic,
        'ic_std': ic_std,
        'economic_sense': economic_sense,
        'overfitting_test': 'pass' if avg_ic > 0.02 else 'fail'
    }

def esg_score_from_alternative_data(company):
    """
    用另类数据计算ESG评分
    """
    # E: 环境
    carbon_emissions = estimate_carbon_from_satellite(company['factories'])
    deforestation = detect_deforestation(company['supply_chain'])
    
    # S: 社会
    labor_disputes = count_labor_news(company['name'])
    employee_sentiment = analyze_glassdoor_reviews(company['name'])
    
    # G: 治理
    governance_scandals = detect_governance_news(company['name'])
    board_diversity = get_board_diversity(company['ticker'])
    
    esg_score = calculate_esg_score(E, S, G)
    
    return esg_score

# 如果ESG评分上升 → 做多（机构资金流入）
# 如果ESG评分下降 → 做空（ESG基金抛售）

def analyze_whale_transactions(token_symbol, min_transaction_size=1000000):
    """
    分析巨鲸交易（链上数据）
    
    巨鲸：持有大量代币的地址
    逻辑：巨鲸转入交易所 → 可能抛售 → 价格下跌
    """
    # 连接区块链API（如Etherscan, Blockchain.com）
    transactions = get_blockchain_transactions(token_symbol)
    
    # 筛选大额交易
    whale_txns = transactions[transactions['amount'] > min_transaction_size]
    
    # 分析流向
    for txn in whale_txns:
        if txn['to_address'] in EXCHANGE_ADDRESSES:
            # 巨鲸转入交易所 → 可能抛售
            signal = 'SELL'
        elif txn['from_address'] in EXCHANGE_ADDRESSES:
            # 巨鲸从交易所提出 → 可能长期持有
            signal = 'BUY'
        
        return signal