jackpipipi

---
name: news-sentiment-scan
description: 舆情监控与情绪分析技能。扫描港股、美股、A股等公司公告、新闻报道、券商研报、社交媒体（微博、雪球等），去噪后进行情绪打分（-10至+10），输出情绪温度计与重大事件清单。触发场景：舆情监控、情绪分析、新闻情绪、社交媒体情绪、上市公司消息面分析、研报解读、"监控XX股票舆情"、"XX新闻情绪如何"。
---

# News Sentiment Scan

舆情监控与情绪分析技能，扫描多渠道信息源并进行情绪打分。

## 功能

- **多渠道信息采集**：公司公告、新闻报道、券商研报、社交媒体（微博、雪球）
- **去噪处理**：过滤广告、水军、无关信息
- **情绪打分**：-10（极度负面）至 +10（极度正面）
- **事件提取**：识别并量化每条重大事件的情绪贡献
- **输出格式**：情绪温度计 + 重大事件清单

## 情绪评分标准

| 分数 | 情绪标签 | 说明 |
|------|----------|------|
| +8 ~ +10 | 🤩 极度乐观 | 重大利好，股价可能大幅上涨 |
| +5 ~ +8 | 😊 偏正面 | 利好消息，持续关注 |
| +2 ~ +5 | 🙂 轻微正面 | 小幅利好，保持观察 |
| -2 ~ +2 | 😐 中性 | 无明显影响 |
| -5 ~ -2 | 😟 轻微负面 | 小幅利空，注意风险 |
| -8 ~ -5 | 😰 偏负面 | 利空消息，谨慎操作 |
| -10 ~ -8 | 😱 极度悲观 | 重大利空，股价可能大幅下跌 |

## 使用方式

### 基本命令

```bash
python3 {SKILL_DIR}/scripts/sentiment_scan.py <stock_code> [days] [market]
```

**参数说明：**
- `stock_code`：股票代码
  - 港股：XXXX.HK（如 0700.HK）或 5位数字（如 00700）
  - 美股：字母代码（如 AAPL、TSLA）
  - A股：6位数字（如 002594、600519）
- `days`：监控天数，默认 7
- `market`：市场标识（可选，auto 自动识别）

**示例：**
```bash
# 监控比亚迪近7天舆情
python3 {SKILL_DIR}/scripts/sentiment_scan.py 002594 7

# 监控腾讯近30天舆情
python3 {SKILL_DIR}/scripts/sentiment_scan.py 0700.HK 30

# 监控苹果近7天舆情
python3 {SKILL_DIR}/scripts/sentiment_scan.py AAPL 7 us
```

## 输出格式

```
==================================================
📰 舆情监控报告：{股票名称}（{代码}）
==================================================

📅 监控周期：最近{N}天
🔍 信息来源：{来源列表}
⏰ 生成时间：{时间戳}

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

🌡️ 情绪温度计：{分数}（{情绪标签}）

   -10  ════════════●════════════  +10
        😱        😐        🤩
        └──────────┘
            {当前情绪位置}

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

📋 重大事件清单：

1. [{情绪图标}] {事件标题}
   来源：{来源} | 时间：{日期}
   情绪贡献：{贡献分数} | 置信度：{置信度}%
   
2. [{情绪图标}] {事件标题}
   来源：{来源} | 时间：{日期}
   情绪贡献：{贡献分数} | 置信度：{置信度}%
   
...

━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

📊 情绪统计：
├── 正面事件：{N}条（{占比}%）
├── 中性事件：{N}条（{占比}%）
├── 负面事件：{N}条（{占比}%）
└── 平均情绪：{平均分}

💡 操作建议：{建议}
```

## 情绪来源权重

| 来源 | 权重 | 说明 |
|------|------|------|
| 公司公告 | 1.0 | 官方信源，最权威 |
| 券商研报 | 0.9 | 专业机构分析 |
| 主流媒体 | 0.8 | 财经媒体新闻 |
| 监管文件 | 0.9 | 交易所、证监会文件 |
| 社交媒体 | 0.6 | 微博、雪球等 |
| 论坛帖子 | 0.5 | 股吧、社区讨论 |

## 事件类型与情绪贡献

| 事件类型 | 基础分数 | 说明 |
|----------|----------|------|
| 业绩超预期 | +5 | 营收/利润大幅增长 |
| 研报上调评级 | +3 | 券商目标价上调 |
| 大股东增持 | +3 | 管理层看好 |
| 政策利好 | +4 | 行业政策支持 |
| 产品发布 | +2 | 新产品上市 |
| 业绩不及预期 | -4 | 营收/利润下降 |
| 研报下调评级 | -3 | 券商目标价下调 |
| 大股东减持 | -3 | 管理层不看好 |
| 政策利空 | -4 | 行业政策收紧 |
| 监管调查 | -5 | 证监会/交易所调查 |

## 数据来源

1. **Yahoo Finance**：公司新闻、分析师评级
2. **Google News**：财经媒体报道
3. **Twitter/X**：社交媒体讨论
4. **微博**：中国社交媒体（需要网络访问）
5. **雪球**：投资社区讨论

## 注意事项

- 情绪分析基于自然语言处理，可能存在误差
- 重大事件需结合基本面判断
- 本工具仅供参考，不构成投资建议
- 社交媒体信息可能被操纵，请谨慎判断

FILE:scripts/sentiment_scan.py
#!/usr/bin/env python3
"""
舆情监控与情绪分析脚本
扫描新闻、公告、社交媒体，进行情绪打分
"""

import sys
import json
import re
from datetime import datetime, timedelta
from collections import defaultdict

try:
    import yfinance as yf
    import pandas as pd
except ImportError as e:
    print(json.dumps({"status": "error", "message": f"缺少依赖: {e}，请运行: pip install yfinance pandas"}))
    sys.exit(1)


# ==================== 情绪词典 ====================

POSITIVE_WORDS = [
    "超预期", "大涨", "涨停", "突破", "增长", "盈利", "买入", "推荐", "增持", "上调",
    "看好", "强劲", "创新高", "首涨", "领涨", "利好", "业绩", "高增长", "超预期",
    "upgrade", "bullish", "beat", "surge", "jump", "rally", "buy", "recommend",
    "strong", "growth", "profit", "soar", "rally", "gain", "positive", "upside"
]

NEGATIVE_WORDS = [
    "亏损", "大跌", "跌停", "下调", "减持", "卖出", "预警", "风险", "调查", "利空",
    "造假", "欺诈", "违规", "处罚", "下修", "不及预期", "暴跌", "破发", "警示",
    "downgrade", "bearish", "miss", "plunge", "crash", "sell", "warning", "risk",
    "investigate", "fraud", "penalty", "loss", "decline", "drop", "fall", "negative"
]

EVENT_PATTERNS = {
    # 业绩相关
    r"(业绩|营收|利润|净利润).*(增长|大涨|超预期|提升|创新高)": ("业绩超预期", 5),
    r"(业绩|营收|利润).*(下降|下滑|减少|不及预期|亏损)": ("业绩不及预期", -4),
    r"Q[1-4].*(营收|利润|同比).*(\d+%)": ("季度业绩公告", 2),
    
    # 评级相关
    r"(买入|推荐|增持|强烈推荐|上调).*(评级|目标价|评级)": ("券商上调评级", 3),
    r"(卖出|减持|中性|下调).*(评级|目标价|评级)": ("券商下调评级", -3),
    r"券商|投行|机构.*(研报|评级|目标价)": ("券商研报发布", 2),
    
    # 股东相关
    r"(大股东|高管|管理层|董事长).*(增持|买入|回购)": ("大股东增持", 3),
    r"(大股东|高管|管理层|董事长).*(减持|卖出|抛售)": ("大股东减持", -3),
    
    # 政策相关
    r"(政策|补贴|支持|利好).*(出台|发布|加大)": ("政策利好", 4),
    r"(监管|政策|限制|收紧).*(出台|发布|加强)": ("监管政策收紧", -3),
    
    # 产品相关
    r"(新品|新产品|新产品).*(发布|上市|推出)": ("产品发布", 2),
    r"(召回|质量问题|投诉).*(产品)": ("产品质量问题", -3),
    
    # 合作相关
    r"(合作|签约|战略|伙伴).*(达成|签署|签订)": ("战略合作", 2),
    r"(终止|取消|破裂|解约)": ("合作终止", -2),
    
    # 诉讼相关
    r"(诉讼|起诉|仲裁|纠纷)": ("法律诉讼", -2),
    r"(罚款|处罚|警示函)": ("监管处罚", -4),
}


# ==================== 工具函数 ====================

def get_sentiment_label(score):
    """根据分数返回情绪标签"""
    if score >= 8:
        return "🤩 极度乐观"
    elif score >= 5:
        return "😊 偏正面"
    elif score >= 2:
        return "🙂 轻微正面"
    elif score >= -2:
        return "😐 中性"
    elif score >= -5:
        return "😟 轻微负面"
    elif score >= -8:
        return "😰 偏负面"
    else:
        return "😱 极度悲观"


def get_sentiment_icon(score):
    """获取情绪图标"""
    if score >= 5:
        return "🟢"
    elif score >= 2:
        return "🟡"
    elif score >= -2:
        return "⚪"
    elif score >= -5:
        return "🟠"
    else:
        return "🔴"


def calculate_sentiment(text, source_weight=0.8):
    """
    计算文本情绪分数
    返回: (分数, 事件类型, 置信度)
    """
    if not text:
        return 0, None, 0
    
    text_lower = text.lower()
    score = 0
    matched_events = []
    
    # 检查事件模式
    for pattern, (event_type, event_score) in EVENT_PATTERNS.items():
        if re.search(pattern, text, re.IGNORECASE):
            matched_events.append(event_type)
            score += event_score
    
    # 词汇情绪分析
    pos_count = sum(1 for word in POSITIVE_WORDS if word.lower() in text_lower)
    neg_count = sum(1 for word in NEGATIVE_WORDS if word.lower() in text_lower)
    
    word_sentiment = (pos_count - neg_count) * 0.5
    score += word_sentiment
    
    # 归一化到 -10 到 +10
    score = max(-10, min(10, score))
    
    # 计算置信度
    confidence = min(100, (abs(score) * 10 + len(matched_events) * 15))
    
    return round(score, 1), matched_events[0] if matched_events else None, round(confidence, 1)


def normalize_ticker(ticker, market=None):
    """标准化股票代码"""
    ticker = ticker.strip()
    
    if market == "hk" or (not market and ticker.endswith(".HK")):
        if not ticker.endswith(".HK"):
            ticker = ticker.zfill(5) + ".HK"
        return ticker.upper(), "hk"
    
    if market == "us" or (not market and ticker.isalpha()):
        return ticker.upper(), "us"
    
    if market == "a" or (not market and ticker.endswith(".SZ") or ticker.endswith(".SS")):
        return ticker, "a"
    
    # 自动识别
    if ticker.endswith(".HK"):
        return ticker.upper(), "hk"
    elif ticker.endswith(".SS"):
        return ticker, "a"
    elif ticker.endswith(".SZ"):
        return ticker, "a"
    elif ticker.isdigit():
        if len(ticker) == 5:
            return ticker.zfill(5) + ".HK", "hk"
        elif len(ticker) == 6:
            if ticker.startswith(("0", "3")):
                return ticker + ".SZ", "a"
            else:
                return ticker + ".SS", "a"
    
    return ticker.upper(), "us"


def get_company_news(ticker, days=7):
    """获取公司新闻"""
    try:
        ticker_obj = yf.Ticker(ticker)
        company_info = ticker_obj.info
        
        news = []
        
        # 获取新闻
        if hasattr(ticker_obj, 'news') and ticker_obj.news:
            for item in ticker_obj.news:
                news.append({
                    "title": item.get("title", ""),
                    "publisher": item.get("publisher", "Unknown"),
                    "link": item.get("link", ""),
                    "published": item.get("published", ""),
                    "type": "news"
                })
        
        return news, company_info
    except Exception as e:
        return [], {}


def analyze_news_sentiment(news_list, days=7):
    """分析新闻情绪"""
    cutoff_date = datetime.now() - timedelta(days=days)
    events = []
    
    for item in news_list:
        title = item.get("title", "")
        if not title:
            continue
        
        # 计算情绪
        score, event_type, confidence = calculate_sentiment(title)
        
        if abs(score) >= 1:  # 只保留有意义的情绪
            events.append({
                "title": title,
                "source": item.get("publisher", "Unknown"),
                "time": item.get("published", ""),
                "score": score,
                "event_type": event_type,
                "confidence": confidence,
                "type": item.get("type", "news")
            })
    
    # 按分数排序
    events.sort(key=lambda x: abs(x["score"]), reverse=True)
    
    return events


def generate_report(ticker, company_name, events, days):
    """生成舆情报告"""
    if not events:
        return {
            "status": "success",
            "ticker": ticker,
            "company_name": company_name,
            "period_days": days,
            "sentiment_score": 0,
            "sentiment_label": "😐 中性",
            "events": [],
            "summary": {
                "positive_count": 0,
                "neutral_count": 0,
                "negative_count": 0
            },
            "message": "未找到相关新闻或公告"
        }
    
    # 计算总体情绪
    total_score = sum(e["score"] for e in events)
    avg_score = total_score / len(events) if events else 0
    
    # 分类统计
    positive = [e for e in events if e["score"] > 2]
    neutral = [e for e in events if -2 <= e["score"] <= 2]
    negative = [e for e in events if e["score"] < -2]
    
    # 生成建议
    if avg_score >= 5:
        suggestion = "整体情绪偏正面，可适当关注，但需结合基本面谨慎操作"
    elif avg_score >= 2:
        suggestion = "情绪整体偏暖，可保持关注，注意观察后续催化剂"
    elif avg_score >= -2:
        suggestion = "情绪中性，建议观望，等待更明确的信号"
    elif avg_score >= -5:
        suggestion = "情绪偏弱，注意风险，建议谨慎操作"
    else:
        suggestion = "情绪明显负面，短期规避，等待市场情绪改善"
    
    return {
        "status": "success",
        "ticker": ticker,
        "company_name": company_name,
        "period_days": days,
        "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "sentiment_score": round(avg_score, 1),
        "sentiment_label": get_sentiment_label(avg_score),
        "events": events[:10],  # 最多10个重大事件
        "summary": {
            "total_events": len(events),
            "positive_count": len(positive),
            "neutral_count": len(neutral),
            "negative_count": len(negative),
            "positive_ratio": round(len(positive) / len(events) * 100, 1) if events else 0,
            "negative_ratio": round(len(negative) / len(events) * 100, 1) if events else 0
        },
        "suggestion": suggestion
    }


def print_report(report):
    """格式化打印报告"""
    if report["status"] != "success":
        print(f"❌ 错误: {report.get('message', '未知错误')}")
        return
    
    ticker = report["ticker"]
    name = report["company_name"]
    days = report["period_days"]
    score = report["sentiment_score"]
    label = report["sentiment_label"]
    events = report.get("events", [])
    summary = report.get("summary", {})
    
    print("=" * 60)
    print(f"📰 舆情监控报告：{name}（{ticker}）")
    print("=" * 60)
    print(f"📅 监控周期：最近{days}天")
    print(f"⏰ 生成时间：{report.get('generated_at', 'N/A')}")
    print()
    
    # 情绪温度计
    print("━" * 60)
    print(f"🌡️ 情绪温度计：{score}（{label}）")
    print()
    
    # 温度计可视化
    bar_length = 40
    # 将 -10 到 +10 映射到 0-40
    position = int((score + 10) / 20 * bar_length)
    
    thermometer = ""
    for i in range(bar_length + 1):
        if i < position:
            thermometer += "█"
        elif i == position:
            thermometer += "●"
        else:
            thermometer += "░"
    
    print(f"   -10 {thermometer} +10")
    print(f"        😱            😐            🤩")
    print("━" * 60)
    print()
    
    # 重大事件清单
    if events:
        print("📋 重大事件清单：")
        print()
        for i, event in enumerate(events, 1):
            icon = get_sentiment_icon(event["score"])
            event_type = event.get("event_type", "一般新闻")
            print(f"{i}. [{icon}] {event['title']}")
            print(f"   来源：{event['source']} | 类型：{event_type}")
            print(f"   情绪贡献：{event['score']:+} | 置信度：{event['confidence']}%")
            print()
    else:
        print("📋 重大事件：无")
        print()
    
    # 情绪统计
    print("━" * 60)
    print("📊 情绪统计：")
    print(f"├── 正面事件：{summary.get('positive_count', 0)}条（{summary.get('positive_ratio', 0)}%）")
    print(f"├── 中性事件：{summary.get('neutral_count', 0)}条")
    print(f"├── 负面事件：{summary.get('negative_count', 0)}条（{summary.get('negative_ratio', 0)}%）")
    print(f"└── 事件总数：{summary.get('total_events', 0)}条")
    print()
    
    # 操作建议
    print("━" * 60)
    print(f"💡 操作建议：{report.get('suggestion', '暂无建议')}")
    print("=" * 60)


# ==================== 主程序 ====================

def main():
    if len(sys.argv) < 2:
        print("用法: python3 sentiment_scan.py <stock_code> [days] [market]")
        print("示例: python3 sentiment_scan.py 002594 7")
        print("      python3 sentiment_scan.py 0700.HK 30")
        print("      python3 sentiment_scan.py AAPL 7 us")
        sys.exit(1)
    
    ticker = sys.argv[1]
    days = int(sys.argv[2]) if len(sys.argv) > 2 else 7
    market = sys.argv[3] if len(sys.argv) > 3 else None
    
    # 输入安全校验
    if not re.match(r'^[A-Za-z0-9.]{1,20}$', ticker):
        print("❌ 无效的股票代码")
        sys.exit(1)
    
    # 标准化代码
    ticker_norm, market_type = normalize_ticker(ticker, market)
    
    print(f"🔍 正在获取 {ticker_norm} 的舆情数据...", file=sys.stderr)
    
    # 获取新闻
    news_list, company_info = get_company_news(ticker_norm, days)
    
    # 分析情绪
    events = analyze_news_sentiment(news_list, days)
    
    # 生成报告
    company_name = company_info.get("shortName", ticker_norm)
    report = generate_report(ticker_norm, company_name, events, days)
    
    # 打印报告
    print_report(report)


if __name__ == "__main__":
    main()