molezzz A股分析全能 Skill（AKShare）
目标：在 OpenClaw 中通过自然语言触发 A 股和相关市场分析，输出适配 QQ/Telegram 的紧凑文本。
- 运行环境：Mac + Python 3.9
- akshare 路径：
/Users/molezz/Library/Python/3.9/lib/python3.9/site-packages - Skill 入口建议：
python3 skills/akshare-stock/main.py --query "${USER_QUERY}"
1) 整体架构设计
采用 Router -> Service -> Analyzer -> Formatter 四层结构，便于扩展和维护。
A. 目录组织（建议）
skills/akshare-stock/
SKILL.md
main.py # OpenClaw 调用入口
router.py # 意图识别 + 参数解析
schemas.py # 数据结构定义（dataclass）
formatter.py # QQ/Telegram 输出模板
services/
market_service.py # 大盘/个股行情、K线、分时、涨跌停、资金流
fundamental_service.py# 财务指标、财报、融资融券、龙虎榜
sector_service.py # 行业/概念板块、轮动、板块资金流
cross_service.py # 期货/期权、基金、可转债、港股/美股
analyzers/
kline_analyzer.py # 均线、振幅、涨跌幅、量比等
flow_analyzer.py # 主力净流入、连续性、强弱排序
rotation_analyzer.py # 板块轮动强度、持续性
adapters/
akshare_adapter.py # 封装 akshare 接口，隔离 API 变化
utils/
trading_calendar.py # 交易日判断
symbols.py # 指数/股票/板块别名映射
cache.py # 短缓存（30~120 秒）
B. 核心流程
main.py接收自然语言 query。router.py输出结构化意图：intent + symbols + timeframe + metric + top_n。services/*拉取原始数据（只做数据获取和轻清洗）。analyzers/*做指标计算和结论生成。formatter.py按聊天平台压缩输出（短句、分段、emoji、重点数值）。
C. 关键设计点
- 意图优先级：先识别“任务类型”，再解析标的和参数，避免误判。
- 适配层隔离：akshare 接口若改名，只需改
adapters/akshare_adapter.py。 - 容错回退：实时接口失败时回退到最近交易日数据，并标注“非实时”。
- 缓存策略：
- 大盘/资金流：30~60 秒
- 板块排行：60~120 秒
- 财报/财务：当天缓存
- 消息长度控制：单条建议 <= 1000 字符；超长自动拆分 2~3 条。
2) 触发词设计（自然语言路由）
建议采用“关键词 + 正则 + 别名词典”混合方式。
A. 意图分类（Intent）
INDEX_REALTIME：实时大盘KLINE_ANALYSIS：历史 K 线INTRADAY_ANALYSIS：分时分析LIMIT_STATS：涨跌停统计MONEY_FLOW：资金流向FUNDAMENTAL：财务指标 / 财报MARGIN_LHB：融资融券 / 龙虎榜SECTOR_ANALYSIS：行业/概念/轮动/板块资金DERIVATIVES：期货/期权FUND_BOND：基金净值 / 可转债HK_US_MARKET：港股 / 美股
B. 触发词样例
- 实时大盘：
A股大盘上证现在多少沪深300实时 - K线：
贵州茅台近60日K线宁德时代周线比亚迪月线复权 - 分时：
看下000001分时平安银行今天分时走势 - 涨跌停：
今日涨停统计跌停家数连板梯队 - 资金流：
主力资金流入前十北向资金行业资金净流入 - 基本面：
茅台财务指标宁德时代最新季报ROE和毛利率 - 融资融券/龙虎榜：
中兴通讯融资融券今日龙虎榜 - 板块：
行业板块涨幅榜概念轮动AI板块资金流 - 其他市场：
IF主力合约300ETF期权基金净值可转债行情腾讯港股英伟达美股
C. 参数抽取规则
- 股票代码：
\b\d{6}\b（如600519） - 日期：
YYYYMMDD/YYYY-MM-DD/今天/昨日/近N日 - 周期：
1m/5m/15m/30m/60m/day/week/month - 排名：
前N（默认 10） - 复权：
前复权/后复权/不复权
3) 各功能实现思路
下面是“功能 -> 推荐数据 -> 分析输出”的落地框架（接口以 akshare 当前版本为准，实际以 adapter 层统一封装）。
3.1 实时大盘行情（已有基础版，升级点）
- 数据：上证、深成指、创业板、沪深300、上证50、科创50。
- 增强：加入成交额、振幅、领涨板块、北向资金当日净流入。
- 输出：
指数点位 + 涨跌幅 + 市场情绪一句话。
3.2 行情分析
- 历史K线：
- 数据：日/周/月 K 线（复权可选）。
- 指标：近 N 日涨跌幅、5/10/20 日均线、量能变化、波动率。
- 输出：趋势判断（多头/震荡/走弱）+ 关键位（支撑/压力）。
- 分时数据：
- 数据：分钟级行情。
- 指标：VWAP 偏离、盘中高低点、午后资金回流。
- 涨跌停统计：
- 数据：涨停池、跌停池、连板梯队。
- 指标：涨停家数、炸板率、最高连板、情绪评分。
- 资金流向：
- 数据：个股/行业/市场资金流。
- 指标：主力净流入 TopN、连续净流入天数、资金集中度。
3.3 基本面分析
- 个股财务指标：ROE、毛利率、净利率、资产负债率、经营现金流。
- 财报数据：营收同比、净利润同比、扣非净利润同比、EPS。
- 融资融券：融资余额、融券余额、日变动，识别杠杆偏好。
- 龙虎榜：上榜原因、买卖前五席位净额、游资活跃度。
- 输出风格：
核心指标摘要 + 同比/环比 + 风险提示。
3.4 板块分析
- 行业板块涨跌：行业涨跌幅榜、成交额、上涨家数。
- 概念板块轮动：近 5 日强度、持续性、日内切换速度。
- 板块资金流向：行业/概念净流入排行 + 领涨龙头。
- 输出：
强势板块Top3 + 轮动结论 + 次日观察点。
3.5 其他（跨市场）
- 期货/期权：主力合约价格、涨跌、持仓变化；期权 PCR（若可得）。
- 基金净值：开放式基金净值、估值偏离、近一周收益。
- 可转债：价格、溢价率、正股联动、成交额。
- 港股/美股：实时行情、近5日表现、与A股联动提示。
4) 代码示例框架（骨架）
说明：以下为可直接落地的最小框架，不含完整业务细节。
main.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
from router import parse_query
from services.market_service import MarketService
from services.fundamental_service import FundamentalService
from services.sector_service import SectorService
from services.cross_service import CrossService
from formatter import render_output
def dispatch(intent_obj):
intent = intent_obj.intent
if intent in {"INDEX_REALTIME", "KLINE_ANALYSIS", "INTRADAY_ANALYSIS", "LIMIT_STATS", "MONEY_FLOW"}:
data = MarketService().handle(intent_obj)
elif intent in {"FUNDAMENTAL", "MARGIN_LHB"}:
data = FundamentalService().handle(intent_obj)
elif intent == "SECTOR_ANALYSIS":
data = SectorService().handle(intent_obj)
elif intent in {"DERIVATIVES", "FUND_BOND", "HK_US_MARKET"}:
data = CrossService().handle(intent_obj)
else:
data = {"ok": False, "error": "未识别请求，请补充标的或时间范围"}
return data
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--query", required=True, help="自然语言请求")
parser.add_argument("--platform", default="qq", choices=["qq", "telegram"])
args = parser.parse_args()
intent_obj = parse_query(args.query)
result = dispatch(intent_obj)
text = render_output(intent_obj, result, platform=args.platform)
print(text)
if __name__ == "__main__":
main()
router.py
from dataclasses import dataclass, field
import re
@dataclass
class IntentObj:
intent: str
symbols: list = field(default_factory=list)
timeframe: str = "day"
days: int = 60
top_n: int = 10
date: str = ""
raw_query: str = ""
def parse_query(query: str) -> IntentObj:
q = query.strip()
obj = IntentObj(intent="INDEX_REALTIME", raw_query=q)
# 1) intent
if any(k in q for k in ["K线", "日线", "周线", "月线"]):
obj.intent = "KLINE_ANALYSIS"
elif "分时" in q:
obj.intent = "INTRADAY_ANALYSIS"
elif any(k in q for k in ["涨停", "跌停", "连板"]):
obj.intent = "LIMIT_STATS"
elif "资金" in q:
obj.intent = "MONEY_FLOW"
elif any(k in q for k in ["财务", "财报", "ROE", "毛利率"]):
obj.intent = "FUNDAMENTAL"
elif any(k in q for k in ["融资融券", "龙虎榜"]):
obj.intent = "MARGIN_LHB"
elif any(k in q for k in ["板块", "行业", "概念", "轮动"]):
obj.intent = "SECTOR_ANALYSIS"
elif any(k in q for k in ["期货", "期权"]):
obj.intent = "DERIVATIVES"
elif any(k in q for k in ["基金", "净值", "可转债"]):
obj.intent = "FUND_BOND"
elif any(k in q for k in ["港股", "美股", "纳斯达克", "道琼斯"]):
obj.intent = "HK_US_MARKET"
# 2) symbol
code_hits = re.findall(r"\b\d{6}\b", q)
if code_hits:
obj.symbols = code_hits
# 3) topN
m = re.search(r"前\s*(\d+)", q)
if m:
obj.top_n = int(m.group(1))
return obj
adapters/akshare_adapter.py
import akshare as ak
class AkAdapter:
def index_spot(self):
return ak.stock_zh_index_spot_sina()
def stock_kline(self, symbol: str, period: str = "daily", start_date: str = "", end_date: str = "", adjust: str = "qfq"):
# 实际参数与函数名按本地 akshare 版本适配
return ak.stock_zh_a_hist(symbol=symbol, period=period, start_date=start_date, end_date=end_date, adjust=adjust)
def stock_intraday(self, symbol: str, period: str = "1"):
return ak.stock_zh_a_minute(symbol=symbol, period=period)
def limit_up_pool(self, date: str):
return ak.stock_zt_pool_em(date=date)
def limit_down_pool(self, date: str):
return ak.stock_dt_pool_em(date=date)
formatter.py
from datetime import datetime
def render_output(intent_obj, result: dict, platform: str = "qq") -> str:
ts = datetime.now().strftime("%Y-%m-%d %H:%M")
if not result.get("ok", False):
return f"⚠️ 请求失败\n原因: {result.get('error', '未知错误')}\n时间: {ts}"
title = result.get("title", "A股分析")
lines = result.get("lines", [])
tips = result.get("tips", "")
# QQ/Telegram 友好输出：短行 + 分段 + 关键数字优先
text = [f"📊 {title}", f"🕒 {ts}", ""]
text.extend(lines[:15])
if tips:
text.extend(["", f"💡 {tips}"])
text.append("\n数据源: akshare")
# 长度保护
merged = "\n".join(text)
return merged[:1000]
输出模板建议（QQ/Telegram）
建议统一为三段：结论 -> 关键数据 -> 风险提示。
示例：
📊 A股午盘情绪
🕒 2026-02-18 11:31
- 上证指数 3210.35（+0.62%）
- 两市成交额 6821 亿，较昨日同期 +8.4%
- 涨停 52 / 跌停 7，连板高度 4
- 主力净流入前三：证券、AI算力、汽车零部件
💡 结论：指数偏强，情绪修复中；但午后关注高位分歧。
数据源: akshare
落地顺序（建议）
- 保留现有实时大盘，抽象进
MarketService.index_realtime()。 - 先补齐行情分析四件套：K线/分时/涨跌停/资金流。
- 再加基本面与板块分析（中频请求，缓存收益高）。
- 最后接入期货/期权/基金/可转债/港美股。
- 每个模块都先做“可读文本输出”，再逐步增加指标深度。
该设计能保证你先快速可用，再逐步增强，不会一次性堆太多接口导致维护困难。