基于华为云MaaS与AI开发者空间的A股智能复盘系统实践

一、为什么需要自动复盘

做 A 股交易的人都知道，收盘后的复盘是最耗时也最容易偷懒的环节。一个完整的复盘流程至少要做四件事：看指数和板块的整体涨跌结构、筛出当日符合自己策略的标的、对持仓股逐一判断是否需要调整、把结论落成可回溯的记录。

手工做这件事有两个硬伤：

• 耗时长且不可持续。沪深两市五千多只标的，哪怕是粗筛，人工也要一两个小时。行情好、情绪高的时候人反而更不愿意做枯燥的记录工作。
• 主观偏差难以避免。连续盈利后容易放大风险偏好，连续回撤后又容易过度保守。复盘如果只靠"凭感觉"，就失去了它本来该有的纠偏作用。

把复盘流程工程化，用程序完成"数据获取—指标计算—策略筛选—结构化研判—报告归档"，既能保证每天都在做，也能让判断标准保持一致。本文记录我在华为云 AI 开发者空间上搭建这套系统的完整过程。

二、整体架构

系统分为五层，每一层都可独立替换：

┌─────────────┐
│  行情接入层   │  富途 OpenAPI / 公开行情接口
└──────┬──────┘
       ▼
┌─────────────┐
│  指标计算层   │  均线、量能、资金流、波动率
└──────┬──────┘
       ▼
┌─────────────┐
│  策略筛选层   │  多策略并行打分，取交集
└──────┬──────┘
       ▼
┌─────────────┐
│ 大模型研判层  │  华为云 MaaS 大模型，输出结构化结论
└──────┬──────┘
       ▼
┌─────────────┐
│  报告生成层   │  Markdown → HTML → PDF 归档
└─────────────┘

关键设计原则：任何一层失败都不能中断整条链路。行情接口超时就降级到备用源，大模型调用失败就只输出数据层结论，保证每天都有报告产出。

三、行情数据接入

以富途 OpenAPI 为例，获取日线 K 线并做前复权处理：

from futu import OpenQuoteContext, KLType, AuType, RET_OK

def fetch_daily(symbols):
    """获取日线数据，返回 {code: DataFrame}"""
    quote_ctx = OpenQuoteContext(host="127.0.0.1", port=11111)
    result = {}
    for code in symbols:
        ret, data, _ = quote_ctx.request_history_kline(
            code, start=None, end=None,
            ktype=KLType.K_DAY,
            autype=AuType.QFQ,      # 前复权，务必统一
            max_count=250,
        )
        if ret != RET_OK:
            print(f"[WARN] {code} 获取失败: {data}")
            continue
        result[code] = data
    quote_ctx.close()
    return result

踩过的坑：

1. 复权方式必须全局统一。前复权（QFQ）适合看技术形态，后复权适合算真实收益，混用会导致均线系统整体失真。
2. 停牌和次新股要单独过滤。上市不足 60 个交易日的标的，多数技术指标还没有参考价值。
3. 时区问题。request_history_kline 返回的时间戳是港交所时区，做日期对齐时要显式转换。

四、核心指标与多策略筛选

我用了三组互相独立的策略，避免单一策略在特定行情下集体失效：

import pandas as pd
import numpy as np

def calc_indicators(df):
    df = df.copy()
    df["ma5"]  = df["close"].rolling(5).mean()
    df["ma20"] = df["close"].rolling(20).mean()
    df["ma60"] = df["close"].rolling(60).mean()
    df["vol_ma5"] = df["volume"].rolling(5).mean()
    # 量能比：当日成交量 / 近 20 日均量
    df["vol_ratio"] = df["volume"] / df["volume"].rolling(20).mean()
    # 20 日波动率
    df["volatility"] = df["close"].pct_change().rolling(20).std() * np.sqrt(250)
    return df

def strategy_trend(df):
    """趋势策略：均线多头排列 + 放量"""
    last = df.iloc[-1]
    return (last["ma5"] > last["ma20"] > last["ma60"]
            and last["vol_ratio"] > 1.5
            and last["close"] > last["ma20"])

def strategy_pullback(df):
    """回调策略：中期趋势向上，短期回踩均线"""
    last = df.iloc[-1]
    prev = df.iloc[-2]
    return (last["ma20"] > last["ma60"]
            and prev["close"] < prev["ma20"]
            and last["close"] > last["ma20"])

def strategy_lowvol(df):
    """低波策略：波动收缩后放量突破"""
    last = df.iloc[-1]
    vol_now = last["volatility"]
    vol_hist = df["volatility"].tail(60).mean()
    return vol_now < vol_hist * 0.7 and last["vol_ratio"] > 2.0

三组策略并行打分，取并集后再按波动率排序，输出候选池。注意：策略逻辑要写成纯函数，输入是 DataFrame，输出是布尔值，这样才好做回测和单元测试。

五、接入大模型做结构化研判

数据层只能回答"哪些标的技术形态符合条件"，回答不了"为什么"。这一层用大模型补齐。华为云 MaaS 提供了 OpenAI 兼容的接口，接入成本很低：

import os, json
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["HWCLOUD_MAAS_KEY"],
    base_url="https://api.modelarts-maas.com/v1",   # 以控制台实际地址为准
)

SYSTEM_PROMPT = """你是严谨的A股复盘助手。只依据给定的量化数据做分析，禁止臆造数据。
输出必须是合法 JSON，字段：
- summary: 一句话结论（不超过50字）
- trend: 趋势判断，取值 up / down / sideways
- risk: 风险提示数组
- action: 建议动作，取值 hold / reduce / watch
置信度字段 confidence 取值 0-1。"""

def analyze(code, indicators: dict) -> dict:
    user_msg = f"标的代码：{code}\n量化指标：{json.dumps(indicators, ensure_ascii=False)}"
    resp = client.chat.completions.create(
        model="qwen-72b",            # 按实际可用的模型名替换
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_msg},
        ],
        temperature=0.2,             # 复盘要稳定，不要发散
        response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content)

几个实用经验：

• temperature 一定要调低。复盘结论需要稳定复现，0.7 以上的温度会让同一份数据给出自相矛盾的结论。
• 强制 JSON 输出。用 response_format 约束，解析失败时重试一次，两次都失败就降级为"仅数据层结论"，不要让整条链路挂掉。
• Prompt 里明确禁止臆造。大模型很容易把训练时见过的旧信息与当前数据混淆，system prompt 里要写死"只依据给定数据"。
• 结论必须带置信度。低置信度的结果在报告里要显式标注，避免被当成确定性信号。

六、报告生成与归档

复盘报告最终要能留存、能检索。我的做法是 Markdown → HTML → PDF：

import markdown, subprocess

md_text = render_report(candidates, llm_results)

with open("report.md", "w", encoding="utf-8") as f:
    f.write(md_text)

# Markdown 转 HTML，指定中文字体避免乱码
html = markdown.markdown(md_text, extensions=["tables", "fenced_code"])
html_full = f"""<html><head><meta charset="utf-8">
<style>body{{font-family:'PingFang SC',sans-serif;line-height:1.7;}}</style>
</head><body>{html}</body></html>"""
open("report.html", "w", encoding="utf-8").write(html_full)

PDF 环节用 Playwright 渲染（Chromium 对中文字体和 CSS 的支持最完整），比用图形库直接绘制 PDF 的方案稳定得多，能彻底避免中文乱码。

七、落地经验总结

1. 数据准确性优先于策略复杂度。源数据错了，再精巧的策略也只会稳定地亏钱。接入新数据源时，先用三只已知历史数据的标的做对账。
2. 降级设计要提前做。行情源、大模型、PDF 渲染都可能失败，每一层都要有"失败也能出报告"的兜底路径。
3. 回测与实盘要共用同一套代码。我最初把策略逻辑在回测脚本里重写了一遍，结果两边参数不一致，实盘跑了半个月才发现。后来把策略抽成独立模块，回测和实盘都 import 同一份，问题才消失。
4. 报告要能回溯。每天的报告按日期归档，附带当日候选池快照。几个月后回看，能清楚看到哪些判断是事后诸葛、哪些确实有前瞻性。

八、小结

把复盘流程拆成"数据—指标—策略—研判—归档"五层之后，每一层都能独立迭代：换数据源不影响策略，换大模型不影响数据层。真正花时间的其实不是代码，而是把判断标准显性化——一旦能把"我觉得该减仓"写成"ma20 下穿 ma60 且量能比 > 2 时减仓 50%"，复盘才真正开始产生复利。

后续计划把这套系统迁移到华为云 AI 开发者空间做常驻运行，配合函数工作流做定时触发，进一步省掉本地维护成本。

免责声明：本文仅为技术实践记录，所涉策略与代码不构成任何投资建议。市场有风险，决策需独立判断。