周红伟：OpenClaw Skills 股票与加密货币分析（Stock Analysis）技能使用参考手册

元宇宙iwemeta

1686人浏览 · 2026-03-08 16:07:39

元宇宙iwemeta · 2026-03-08 16:07:39 发布

概述

Stock Analysis是一个用于分析股票和加密货币的OpenClaw技能，支持8维度股票分析、加密货币分析、投资组合管理、观察列表、股息分析、热门扫描和谣言扫描功能。该技能整合了Yahoo Finance、CoinGecko、CNN Fear & Greed Index、SEC EDGAR等多个数据源，能够为用户提供清晰的BUY/HOLD/SELL投资建议，并附带详细的支持点和风险提示，帮助用户做出更明智的投资决策。

该技能特别适合两类场景：一类是你要做"快评内容"，另一类是你要做"结构化研究笔记"，把零散信息收敛成可复用的判断。

技能信息

名称：stock-analysis
描述：使用Yahoo Finance数据分析股票和加密货币，支持投资组合管理、带警报的观察列表、股息分析、8维度股票评分、热门趋势检测（Hot Scanner）和谣言/早期信号检测（Rumor Scanner）。可用于股票分析、投资组合跟踪、收益反应分析、加密货币监控、热门股票发现，或在主流新闻之前找到市场谣言。
版本：6.2.0
作者：udiedrichsen
主页：https://finance.yahoo.com
依赖：
- Python 3.10+
- yfinance>=0.2.40
- pandas>=2.0.0
- fear-and-greed>=0.4
- edgartools>=2.0.0
- feedparser>=6.0.0
安装命令：npx clawhub@latest install stock-analysis
支持的命令：
- /stock - 分析股票或加密货币（例如：/stock AAPL）
- /stock_compare - 比较多个代码
- /stock_dividend - 分析股息指标
- /stock_watch - 添加/移除观察列表
- /stock_alerts - 检查触发的警报
- /stock_hot - 查找热门股票和加密货币（Hot Scanner）
- /stock_rumors - 查找早期信号、并购谣言、内幕交易活动（Rumor Scanner）
- /portfolio - 显示投资组合摘要
- /portfolio_add - 向投资组合添加资产

👤 作者：udiedrichsen
🦞 官方地址：https://clawhub.ai/udiedrichsen/stock-analysis
👉 Skills 下载地址：stock-analysis-6.2.0.zip

快速开始

股票分析

## 基本分析
uv run scripts/analyze_stock.py AAPL
## 快速模式（跳过内幕交易和突发新闻）
uv run scripts/analyze_stock.py AAPL --fast
## 比较多个股票
uv run scripts/analyze_stock.py AAPL MSFT GOOGL
## 加密货币分析
uv run scripts/analyze_stock.py BTC-USD ETH-USD

股息分析

## 分析股息
uv run scripts/dividends.py JNJ
## 比较股息股票
uv run scripts/dividends.py JNJ PG KO MCD --output json

观察列表与警报

## 添加到观察列表
uv run scripts/watchlist.py add AAPL
## 带价格目标警报
uv run scripts/watchlist.py add AAPL --target 200
## 带止损警报
uv run scripts/watchlist.py add AAPL --stop 150
## 信号变化警报
uv run scripts/watchlist.py add AAPL --alert-on signal
## 查看观察列表
uv run scripts/watchlist.py list
## 检查触发的警报
uv run scripts/watchlist.py check
## 移除观察列表
uv run scripts/watchlist.py remove AAPL

投资组合管理

## 创建投资组合
uv run scripts/portfolio.py create "Tech Portfolio"
## 添加资产
uv run scripts/portfolio.py add AAPL --quantity 100 --cost 150
uv run scripts/portfolio.py add BTC-USD --quantity 0.5 --cost 40000
## 查看投资组合
uv run scripts/portfolio.py show
## 分析投资组合
uv run scripts/analyze_stock.py --portfolio "Tech Portfolio" --period weekly

谣言扫描

## 查找早期信号、并购谣言、内幕交易活动
python3 scripts/rumor_scanner.py

功能详解

1. 股票分析

股票分析功能提供8维度的全面分析，包括：

收益惊喜（30%权重）：EPS超出/低于预期的情况
基本面（20%权重）：市盈率、利润率、增长率等
分析师情绪（20%权重）：评级和目标价格
历史模式（10%权重）：过去的收益反应
市场环境（10%权重）：VIX指数、SPY/QQQ趋势
行业表现（15%权重）：相对行业强度
动量（15%权重）：RSI、52周区间
情绪（10%权重）：恐惧/贪婪指数、空头头寸、内幕交易

分析结果会生成清晰的BUY/HOLD/SELL建议，并附带支持点和风险提示。

2. 加密货币分析

加密货币分析提供3维度的分析，包括：

市值与分类：加密货币的市值排名和分类
BTC相关性：30天内与比特币的相关性
动量：RSI和价格区间

3. 股息分析

股息分析功能提供以下指标：

股息收益率：年度股息/价格
派息率：股息/EPS
5年增长率：股息的复合年增长率
连续增长年数：连续增加股息的年数
安全评分：0-100的综合评分
收入评级：优秀/良好/中等/较差

4. 投资组合管理

投资组合管理功能允许用户创建和管理多个投资组合，跟踪持仓、盈亏和集中度警告，支持股票和加密货币资产。

5. 观察列表与警报

观察列表功能允许用户添加股票和加密货币，并设置价格目标、止损和信号变化警报，当触发警报时会通知用户。

6. 热门扫描

热门扫描功能聚合多个数据源，实时查找热门股票和加密货币，数据源包括：

CoinGecko（热门加密货币、最大涨跌幅）
Google News（财经和加密货币头条）
Yahoo Finance（涨幅榜、跌幅榜、最活跃股票）
Twitter/X（社交情绪，可选）
Reddit（r/wallstreetbets和r/cryptocurrency，可选）

7. 谣言扫描

谣言扫描功能查找早期信号、并购谣言和内幕交易活动，数据源包括：

Twitter/X："hearing that"、"sources say"等关键词
Google News：并购、内幕交易、分析师升级/降级新闻
异常关键词检测

分析维度

股票（8维度）

维度	权重	描述
收益惊喜	30%	EPS超出/低于预期的情况
基本面	20%	市盈率、利润率、增长率、债务等
分析师情绪	20%	分析师评级和目标价格
历史模式	10%	过去的收益反应模式
市场环境	10%	VIX指数、SPY/QQQ趋势
行业表现	15%	相对行业强度
动量	15%	RSI、52周区间、成交量
情绪	10%	恐惧/贪婪指数、空头头寸、内幕交易、看跌/看涨比率

加密货币（3维度）

市值与分类：加密货币的市值排名和分类（如智能合约L1、DeFi等）
BTC相关性：30天内与比特币的相关性
动量：RSI和价格区间

风险检测

该技能会检测以下风险：

收益前风险：距离收益发布少于14天，预期高波动性
收益后飙升：5天内涨幅超过15%，收益可能已被定价
超买：RSI>70且接近52周高点，高风险入场
风险规避模式：GLD/TLT/UUP同时上涨，资金流向避险资产
地缘政治风险：检测台湾、中国、俄罗斯、中东等地缘政治关键词
突发新闻风险：过去24小时内的危机关键词

带中文注释的核心代码

1. analyze_stock.py（股票分析主脚本核心代码）

#!/usr/bin/env python3
"""
股票分析主脚本，使用Yahoo Finance数据进行8维度股票分析和加密货币分析
"""
import argparse
import asyncio
import json
import sys
import time
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
from typing import Literal, Optional, Tuple, List, Dict
import yfinance as yf
import pandas as pd
from fear_and_greed import FearAndGreed
@dataclass
class StockData:
"""股票数据类，存储从Yahoo Finance获取的所有数据"""
ticker: str
asset_type: Literal["stock", "crypto"]
info: Dict
earnings_history: Optional[pd.DataFrame]
analyst_info: Optional[Dict]
price_history: Optional[pd.DataFrame]
@dataclass
class EarningsSurprise:
"""收益惊喜分析结果"""
score: float
explanation: str
actual_eps: Optional[float]
expected_eps: Optional[float]
surprise_pct: Optional[float]
def fetch_stock_data(ticker: str, verbose: bool = False) -> Optional[StockData]:
"""
从Yahoo Finance获取股票数据，带重试逻辑
Args:
ticker: 股票代码
verbose: 是否打印调试信息
Returns:
StockData对象或None
"""
max_retries = 3
for attempt in range(max_retries):
try:
if verbose:
print(f"正在获取 {ticker} 的数据...", file=sys.stderr)
stock = yf.Ticker(ticker)
info = stock.info
# 检测资产类型
asset_type = "crypto" if ticker.endswith("-USD") else "stock"
# 获取收益历史
earnings_history = stock.earnings if hasattr(stock, 'earnings') else None
# 获取分析师信息
analyst_info = stock.recommendations if hasattr(stock, 'recommendations') else None
# 获取价格历史
price_history = stock.history(period="1y")
return StockData(
ticker=ticker,
asset_type=asset_type,
info=info,
earnings_history=earnings_history,
analyst_info=analyst_info,
price_history=price_history
)
except Exception as e:
if verbose:
print(f"获取 {ticker} 数据失败 (尝试 {attempt+1}/{max_retries}): {e}", file=sys.stderr)
if attempt < max_retries - 1:
wait_time = 2 ** attempt # 指数退避
time.sleep(wait_time)
return None
def analyze_earnings_surprise(data: StockData) -> Optional[EarningsSurprise]:
"""
分析收益惊喜情况
Args:
data: StockData对象
Returns:
EarningsSurprise对象或None
"""
if not data.earnings_history or data.earnings_history.empty:
return None
# 获取最近的收益数据
latest_earnings = data.earnings_history.iloc[-1]
actual_eps = latest_earnings.get("Reported EPS")
expected_eps = latest_earnings.get("EPS Estimate")
if actual_eps is None or expected_eps is None:
return None
# 计算惊喜百分比
surprise_pct = ((actual_eps - expected_eps) / abs(expected_eps)) * 100
# 计算分数
score = surprise_pct / 100 # 标准化分数
# 生成解释
if surprise_pct > 0:
explanation = f"超出预期 {surprise_pct:.1f}% - EPS ${actual_eps:.2f} 对比预期 ${expected_eps:.2f}"
else:
explanation = f"低于预期 {abs(surprise_pct):.1f}% - EPS ${actual_eps:.2f} 对比预期 ${expected_eps:.2f}"
return EarningsSurprise(
score=score,
explanation=explanation,
actual_eps=actual_eps,
expected_eps=expected_eps,
surprise_pct=surprise_pct
)
def main():
"""主函数，处理命令行参数并执行分析"""
parser = argparse.ArgumentParser(description="股票和加密货币分析工具")
parser.add_argument("tickers", nargs="+", help="股票或加密货币代码")
parser.add_argument("--output", choices=["text", "json"], default="text", help="输出格式")
parser.add_argument("--verbose", "-v", action="store_true", help="显示详细信息")
parser.add_argument("--fast", action="store_true", help="快速模式，跳过慢速分析")
parser.add_argument("--no-insider", action="store_true", help="跳过内幕交易分析")
parser.add_argument("--portfolio", help="分析投资组合")
parser.add_argument("--period", choices=["daily", "weekly", "monthly"], default="daily", help="分析周期")
args = parser.parse_args()
# 处理投资组合模式
if args.portfolio:
# 加载投资组合
try:
from portfolio import PortfolioStore
store = PortfolioStore()
portfolio = store.get_portfolio(args.portfolio)
if not portfolio:
print(f"错误：投资组合 '{args.portfolio}' 不存在", file=sys.stderr)
sys.exit(1)
if not portfolio.assets:
print(f"投资组合 '{args.portfolio}' 没有资产", file=sys.stderr)
sys.exit(1)
args.tickers = [asset.ticker for asset in portfolio.assets]
if args.verbose:
print(f"正在分析投资组合: {args.portfolio} ({len(args.tickers)} 个资产)", file=sys.stderr)
except ImportError:
print("错误：无法导入portfolio模块", file=sys.stderr)
sys.exit(1)
# 分析每个代码
results = []
for ticker in args.tickers:
data = fetch_stock_data(ticker.upper(), verbose=args.verbose)
if not data:
print(f"错误：无法获取 {ticker} 的数据", file=sys.stderr)
continue
# 分析各个维度
earnings = analyze_earnings_surprise(data)
# 其他分析函数...
# 合成信号
# signal = synthesize_signal(...)
# results.append(signal)
# 输出结果
if args.output == "json":
print(json.dumps([asdict(r) for r in results], indent=2, default=str))
else:
for signal in results:
print(f"{'='*77}")
print(f"股票分析: {signal.ticker} ({signal.company_name})")
print(f"生成时间: {signal.timestamp}")
print(f"{'='*77}")
print(f"建议: {signal.recommendation} (置信度: {signal.confidence:.1%})")
print("\n支持点:")
for point in signal.supporting_points:
print(f"• {point}")
if signal.caveats:
print("\n风险提示:")
for caveat in signal.caveats:
print(f"• {caveat}")
print(f"\n{'='*77}")
print("免责声明：不构成财务建议。")
print(f"{'='*77}\n")
if __name__ == "__main__":
main()

2. portfolio.py（投资组合管理脚本核心代码）

#!/usr/bin/env python3
"""
投资组合管理脚本
"""
import argparse
import json
import sys
from dataclasses import dataclass, asdict
from datetime import datetime
from pathlib import Path
import yfinance as yf
@dataclass
class Asset:
"""资产类，存储投资组合中的资产信息"""
ticker: str
type: str # "stock" or "crypto"
quantity: float
cost_basis: float
added_at: str
@dataclass
class Portfolio:
"""投资组合类，存储投资组合信息"""
name: str
created_at: str
updated_at: str
assets: list[Asset]
class PortfolioStore:
"""投资组合存储管理类，处理投资组合的持久化"""
def __init__(self, path: Optional[Path] = None):
"""初始化存储路径"""
self.path = path or Path.home() / ".clawdbot" / "skills" / "stock-analysis" / "portfolios.json"
self._data: Optional[Dict] = None
def _load(self) -> Dict:
"""从磁盘加载投资组合数据"""
if self._data is not None:
return self._data
try:
if self.path.exists():
with open(self.path, "r", encoding="utf-8") as f:
self._data = json.load(f)
return self._data
except (json.JSONDecodeError, IOError):
pass
# 如果加载失败，初始化空数据
self._data = {"version": 1, "portfolios": {}}
return self._data
def _save(self) -> None:
"""将投资组合数据保存到磁盘（原子写入）"""
if self._data is None:
return
# 确保目录存在
self.path.parent.mkdir(parents=True, exist_ok=True)
# 原子写入：先写入临时文件，然后重命名
tmp_path = self.path.with_suffix(".tmp")
try:
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(self._data, f, indent=2)
tmp_path.replace(self.path)
except Exception:
if tmp_path.exists():
tmp_path.unlink()
raise
def create_portfolio(self, name: str) -> Portfolio:
"""创建新投资组合"""
data = self._load()
key = name.lower().replace(" ", "-")
if key in data["portfolios"]:
raise ValueError(f"投资组合 '{name}' 已存在")
now = datetime.now().isoformat()
portfolio = {
"name": name,
"created_at": now,
"updated_at": now,
"assets": [],
}
data["portfolios"][key] = portfolio
self._save()
return Portfolio(name=name, created_at=now, updated_at=now, assets=[])
def add_asset(
self,
portfolio_name: str,
ticker: str,
quantity: float,
cost_basis: float,
) -> Asset:
"""向投资组合添加资产"""
data = self._load()
key = portfolio_name.lower().replace(" ", "-")
# 查找投资组合
if key not in data["portfolios"]:
raise ValueError(f"投资组合 '{portfolio_name}' 不存在")
portfolio = data["portfolios"][key]
ticker = ticker.upper()
# 检查资产是否已存在
for asset in portfolio["assets"]:
if asset["ticker"] == ticker:
raise ValueError(f"资产 '{ticker}' 已在投资组合中。使用 'update' 来修改。")
# 验证代码
asset_type = "crypto" if ticker.endswith("-USD") else "stock"
try:
stock = yf.Ticker(ticker)
info = stock.info
if "regularMarketPrice" not in info:
raise ValueError(f"无效代码: {ticker}")
except Exception as e:
raise ValueError(f"无法验证代码 '{ticker}': {e}")
now = datetime.now().isoformat()
asset = {
"ticker": ticker,
"type": asset_type,
"quantity": quantity,
"cost_basis": cost_basis,
"added_at": now,
}
portfolio["assets"].append(asset)
portfolio["updated_at"] = now
self._save()
return Asset(**asset)
def main():
"""主函数，处理命令行参数"""
parser = argparse.ArgumentParser(description="投资组合管理工具")
subparsers = parser.add_subparsers(dest="command", help="命令")
# create
create_parser = subparsers.add_parser("create", help="创建新投资组合")
create_parser.add_argument("name", help="投资组合名称")
# add
add_parser = subparsers.add_parser("add", help="向投资组合添加资产")
add_parser.add_argument("ticker", help="股票/加密货币代码（例如：AAPL, BTC-USD）")
add_parser.add_argument("--quantity", "-q", type=float, required=True, help="数量")
add_parser.add_argument("--cost", "-c", type=float, required=True, help="单位成本")
add_parser.add_argument("--portfolio", "-p", help="投资组合名称（默认：第一个投资组合）")
# 其他命令...
args = parser.parse_args()
if not args.command:
parser.print_help()
sys.exit(1)
store = PortfolioStore()
try:
if args.command == "create":
portfolio = store.create_portfolio(args.name)
print(f"创建投资组合: {portfolio.name}")
elif args.command == "add":
portfolio_name = args.portfolio or store.get_default_portfolio_name()
if not portfolio_name:
print("没有找到投资组合。先使用 'create' 命令创建一个。")
sys.exit(1)
asset = store.add_asset(portfolio_name, args.ticker, args.quantity, args.cost)
print(f"已向 {portfolio_name} 添加 {asset.ticker} ({asset.type}): "
f"{asset.quantity} 单位 @ ${asset.cost_basis:.2f}")
# 其他命令处理...
except ValueError as e:
print(f"错误: {e}")
sys.exit(1)
except Exception as e:
print(f"意外错误: {e}")
sys.exit(1)
if __name__ == "__main__":
main()