akshare-stock

#!/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()

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

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)

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]

📊 A股午盘情绪
🕒 2026-02-18 11:31

- 上证指数 3210.35（+0.62%）
- 两市成交额 6821 亿，较昨日同期 +8.4%
- 涨停 52 / 跌停 7，连板高度 4
- 主力净流入前三：证券、AI算力、汽车零部件

💡 结论：指数偏强，情绪修复中；但午后关注高位分歧。
数据源: akshare