最近在做 AI 量化交易工具时遇到了一个很有意思的需求单一大模型看图分析 K 线结果往往不够稳定。有的模型擅长识别形态有的模型更关注均线关系还有的模型对成交量变化更敏感。于是我们做了一项新功能让多个 AI 同时看同一张图各自独立给出分析最终汇总为一份综合研判结果。今天把这项功能的实现思路、关键代码和踩坑记录完整整理出来希望对你构建自己的 AI 量化分析系统有帮助。本文适合有 Python 基础的开发者涉及的内容包括多模态大模型 API 调用、K 线图表生成、多模型并行分析的任务拆分、结果聚合策略以及在实际工程中容易忽略的稳定性和成本问题。1. 背景与核心概念1.1 单模型看图分析的不足传统量化策略分析师依赖 K 线、均线、MACD、成交量等指标做判断。大模型出现后很多人尝试直接把 K 线截图丢给 GPT-4V、Claude、Gemini 这类多模态模型让它们“看图说话”。单模型分析有一个明显问题不同模型的能力侧重点差异很大。有的模型对图表中的文字、数字比较敏感能准确读出 K 线数据有的模型长于趋势判断但对具体数值容易出错还有的模型在形态识别上表现不错却忽略成交量配合。这意味着单一大模型的分析结果往往带有“盲区”。如果多个 AI 同时看同一张图从不同角度给出各自分析再把这些分析做交叉验证和综合汇总就能显著提升分析的稳定性和参考价值。1.2 多 AI 并行分析的基本思路所谓“多个 AI 同时看一张图”本质上是统一生成一张包含核心行情信息的 K 线图。把这张图同时发给多个多模态 AI 模型。每个模型独立输出对当前行情的分析互不干扰。系统收集所有模型的分析结果按结构要求汇总输出。示意图可以理解为下面的简单流程K线图文件 - 模型A(趋势判断) - 分析A - 模型B(形态识别) - 分析B - 汇总引擎 - 综合研判报告 - 模型C(量价关系) - 分析C这里的核心不是“多个模型聊天”而是彼此隔离的独立分析通道最后再统一汇总。这样设计的好处是避免模型之间的观点互相污染也能保证分析结果可回溯——每一份分析都能找到对应的模型来源。1.3 典型应用场景多 AI 联合看盘功能可以应用在几个实际场景中盘前复盘每天开盘前让多个 AI 对昨日 K 线图做独立复盘生成多视角解读。个股异动分析发现某只股票短期涨幅异常时让多个 AI 分别分析原因和后续风险。策略信号验证量化策略产生买点信号后用多 AI 分析来辅助确认信号质量。学习辅助新手可以把多 AI 的分析结果作为参考理解不同流派对同一张图的不同解释。2. 环境准备与版本说明本文示例代码以 Python 3.10 为基础核心依赖如下pip install akshare pandas matplotlib openai anthropic需要说明的是模型 API 的版本迭代速度很快下面的代码以常见的 OpenAI、Anthropic 接口为例演示整体思路。你在实际项目中要根据自己使用的模型官方文档调整请求参数不要照抄后直接用于生产环境。操作系统以 Windows 10/11 或 Ubuntu 20.04 均可代码不依赖某个特定系统特性。这里用到的模型配置如下模型用途说明模型A趋势与均线判断擅长整体趋势解读模型BK线形态识别擅长识别锤子线、吞没形态等模型C量价关系分析擅长结合成交量判断实际项目中你可以替换为任意支持图像输入的多模态模型。建议至少准备两个不同厂商的模型避免因为单家模型服务异常导致整个功能不可用。3. 核心原理拆解3.1 为什么要把分析角色拆开多 AI 并行分析并不是简单地“让多个模型重复做同一件事”而是为每个模型分配一个明确的分析角色。比如模型A的提示词强调“只关注趋势和均线系统忽略其他细节”。模型B的提示词强调“只识别经典K线形态给出出现位置”。模型C的提示词强调“关注成交量与价格的关系”。这样做的好处有三点减少幻觉。模型在明确的角色约束下不会产生大而全但模糊的回答。提升可读性。汇总结果时每个维度清晰独立。方便交叉验证。当多个模型对同一现象给出相似结论时可信度更高。3.2 图像输入的处理方式大模型读取图片有两种常见方式Base64 编码把图片文件编码为 Base64 字符串放入请求消息中。图片 URL把图片上传到临时图床把 URL 传给模型。本地量化分析场景推荐使用 Base64 方式因为不依赖外网图床更安全。但要注意Base64 会把图片体积膨胀约 33%请求体可能较大需要合理压缩图片大小。3.3 提示词设计框架多 AI 看图分析的提示词建议统一采用以下框架你是[角色]。请分析这张K线图。 重点关注[具体分析维度1]、[具体分析维度2]。 回答要求 1. 用中文回答控制在200字以内。 2. 先给出结论再简要说明依据。 3. 如果图片不清晰或无法判断明确说明。这里的核心是限制回答范围。不要问模型“这张图说明什么”而要问“从趋势角度看这张图呈现了什么状态”。模型在约束条件下输出的内容更可控。4. 完整实战案例构建多 AI 联合看图分析系统下面我们逐步实现一个最小可用的多 AI 看图分析系统。整体分为四个部分使用 akshare 获取行情数据。使用 matplotlib 生成带均线和成交量的 K 线图。调用多个多模态大模型 API 分别分析图片。汇总各模型输出生成综合报告。4.1 创建项目结构首先创建项目目录multi_ai_analysis/ ├── chart_utils.py # 生成K线图 ├── ai_clients.py # 封装多模型调用 ├── aggregator.py # 汇总分析结果 ├── main.py # 主入口 └── config.py # 配置文件4.2 生成 K 线图我们使用 akshare 获取 A 股或指数的日线数据然后用 matplotlib 绘制带均线和成交量的 K 线图。# 文件路径multi_ai_analysis/chart_utils.py # -*- coding: utf-8 -*- 生成K线图包含K线、5日/20日均线、成交量 import akshare as ak import pandas as pd import matplotlib import matplotlib.pyplot as plt from matplotlib.patches import Rectangle matplotlib.use(Agg) # 使用无界面后端适合服务器环境 plt.rcParams[font.sans-serif] [SimHei, Microsoft YaHei, PingFang SC] plt.rcParams[axes.unicode_minus] False def fetch_daily_data(symbol: str, days: int 60) - pd.DataFrame: 获取日线数据返回包含 open/high/low/close/volume 的DataFrame。 symbol 示例sh600519表示上证贵州茅台 # 这里以股票日线为例实际可以根据需要获取指数或期货数据 df ak.stock_zh_a_hist(symbolsymbol, perioddaily, adjustqfq) df df.tail(days).reset_index(dropTrue) df.columns [date, open, close, high, low, volume, amount, amplitude, pct_change, change, turnover] return df def plot_kline(df: pd.DataFrame, save_path: str) - str: 绘制K线图并保存到 save_path。 df[ma5] df[close].rolling(5).mean() df[ma20] df[close].rolling(20).mean() fig, ax plt.subplots(2, 1, figsize(14, 10), gridspec_kw{height_ratios: [3, 1]}) plt.subplots_adjust(hspace0.05) # 绘制K线 for idx, row in df.iterrows(): color red if row[close] row[open] else green # 影线 ax[0].plot([idx, idx], [row[low], row[high]], colorcolor, linewidth0.8) # 实体 body_bottom min(row[open], row[close]) body_height abs(row[close] - row[open]) if body_height 1e-10: body_height 0.01 rect Rectangle((idx - 0.3, body_bottom), 0.6, body_height, facecolorcolor, edgecolorcolor) ax[0].add_patch(rect) # 均线 ax[0].plot(df[ma5], labelMA5, colororange, linewidth1.2) ax[0].plot(df[ma20], labelMA20, colorblue, linewidth1.2) ax[0].legend() ax[0].set_title(KLine Analysis) ax[0].set_ylabel(Price) ax[0].grid(True, alpha0.3) # 成交量 vol_bars ax[1].bar(df.index, df[volume], color[red if c o else green for o, c in zip(df[open], df[close])], alpha0.6) ax[1].set_ylabel(Volume) ax[1].grid(True, alpha0.3) # 标注日期刻度避免太密集 tick_step max(1, len(df) // 8) tick_indices list(range(0, len(df), tick_step)) ax[0].set_xticks(tick_indices) ax[0].set_xticklabels([df.iloc[i][date] for i in tick_indices], rotation30) ax[1].set_xticks(tick_indices) ax[1].set_xticklabels([df.iloc[i][date] for i in tick_indices], rotation30) plt.savefig(save_path, dpi120, bbox_inchestight) plt.close(fig) return save_path这里有几个细节需要解释成交量使用红绿柱表示涨跌与 K 线保持一致。均线只保留 MA5 和 MA20避免图面过于复杂方便大模型识别核心信息。日期刻度做了稀疏处理防止重叠影响模型读取。4.3 封装多模型调用接下来封装 AI 客户端的调用逻辑。不同的模型使用的 SDK 不同我们用一个统一的接口来封装。# 文件路径multi_ai_analysis/ai_clients.py # -*- coding: utf-8 -*- 封装多个多模态大模型的调用接口 import base64 import os from openai import OpenAI import anthropic def _encode_image(image_path: str) - str: 将图片文件编码为 Base64 字符串 with open(image_path, rb) as f: return base64.b64encode(f.read()).decode(utf-8) class BaseAIClient: 统一的AI客户端基类 def analyze(self, image_path: str, prompt: str) - str: raise NotImplementedError class OpenAIClient(BaseAIClient): OpenAI GPT-4o / GPT-4V 客户端示例 def __init__(self, api_key: str None, model: str gpt-4o): self.client OpenAI( api_keyapi_key or os.getenv(OPENAI_API_KEY) ) self.model model def analyze(self, image_path: str, prompt: str) - str: b64_image _encode_image(image_path) response self.client.chat.completions.create( modelself.model, messages[ { role: user, content: [ {type: text, text: prompt}, { type: image_url, image_url: { url: fdata:image/png;base64,{b64_image} } } ] } ], max_tokens500, temperature0.3 ) return response.choices[0].message.content class AnthropicClient(BaseAIClient): Anthropic Claude 客户端示例 def __init__(self, api_key: str None, model: str claude-3-5-sonnet-20241022): self.client anthropic.Anthropic( api_keyapi_key or os.getenv(ANTHROPIC_API_KEY) ) self.model model def analyze(self, image_path: str, prompt: str) - str: b64_image _encode_image(image_path) response self.client.messages.create( modelself.model, max_tokens500, temperature0.3, messages[ { role: user, content: [ { type: image, source: { type: base64, media_type: image/png, data: b64_image } }, {type: text, text: prompt} ] } ] ) return .join(block.text for block in response.content)如果你使用其他模型比如 Google Gemini、智谱 GLM-4V、通义千问 VL 等思路完全一致把图片转为 Base64按照对应模型的请求格式发送即可。4.4 定义各 AI 的分析角色与提示词同一个模型也可以承担不同角色但为了让多 AI 效果更好建议为每个模型分配固定角色。这里我们定义三组提示词# 文件路径multi_ai_analysis/config.py # -*- coding: utf-8 -*- 模型角色与提示词配置 TREND_PROMPT 你是一位趋势分析专家。请分析这张K线图 1. 当前处于上升趋势、下降趋势还是震荡趋势 2. MA5 与 MA20 的位置关系是什么 3. 最近5根K线的整体走势如何 4. 给出未来3-5个交易日的趋势判断偏向谨慎表述。 要求200字以内先给结论再给依据。 PATTERN_PROMPT 你是一位K线形态识别专家。请分析这张K线图 1. 图中是否存在经典K线形态如锤子线、吞没形态、十字星等 2. 如果存在出现在图中什么位置 3. 这些形态在当前位置意味着什么 4. 给出关键支撑位和阻力位的粗略判断。 要求200字以内先给结论再给依据。 VOLUME_PRICE_PROMPT 你是一位量价分析专家。请分析这张K线图 1. 近期成交量与价格走势是否配合 2. 是否出现放量上涨、缩量回调或放量下跌等信号 3. 从量价关系角度看当前多空力量对比如何 4. 给出量价层面的风险提示。 要求200字以内先给结论再给依据。 # 为每个模型指定角色 MODEL_ROLES [ { name: 趋势分析, prompt: TREND_PROMPT, }, { name: 形态识别, prompt: PATTERN_PROMPT, }, { name: 量价分析, prompt: VOLUME_PRICE_PROMPT, }, ]4.5 汇总多模型结果汇总逻辑是整个系统的关键。我们不希望把三个模型的回答简单拼接而是要做结构化整理。# 文件路径multi_ai_analysis/aggregator.py # -*- coding: utf-8 -*- 汇总多个AI的分析结果生成综合报告 import json def aggregate_results(results: list) - dict: results 格式 [ {role_name: 趋势分析, model_name: gpt-4o, content: ...}, {role_name: 形态识别, model_name: claude-..., content: ...}, {role_name: 量价分析, model_name: gpt-4o, content: ...}, ] 返回汇总后的结构化结果 summary { total_models: len(results), success_models: 0, details: [], combined_text: , } for item in results: if item[content] and not item[content].startswith([错误]): summary[success_models] 1 detail { role: item[role_name], model: item[model_name], analysis: item[content], } summary[details].append(detail) # 拼接纯文本方便后续人工阅读或二次处理 parts [] for detail in summary[details]: parts.append(f【{detail[role]}】\n{detail[analysis]}) summary[combined_text] \n\n.join(parts) # 简单的一致性判断如果成功模型少于2个降级提示 if summary[success_models] 2: summary[consensus] 样本不足无法形成有效共识 else: summary[consensus] 多个模型已完成独立分析请结合各维度结论综合判断 return summary def save_report(summary: dict, output_path: str report.json) - None: 保存汇总报告到本地文件 with open(output_path, w, encodingutf-8) as f: json.dump(summary, f, ensure_asciiFalse, indent2)这里做了一个简单的降级处理如果多个模型中只有 1 个返回成功说明系统不稳定不强行生成共识结论。这个设计在实际工程中很重要因为大模型 API 偶尔会超时或返回空内容。4.6 主流程在主流程中串联所有步骤# 文件路径multi_ai_analysis/main.py # -*- coding: utf-8 -*- 主入口多AI联合看图分析 import os from chart_utils import fetch_daily_data, plot_kline from ai_clients import OpenAIClient, AnthropicClient from config import MODEL_ROLES from aggregator import aggregate_results, save_report def build_clients(): 根据环境变量初始化模型客户端。 这里演示同时使用 OpenAI 和 Anthropic 的模型。 clients {} if os.getenv(OPENAI_API_KEY): clients[gpt-4o] OpenAIClient() if os.getenv(ANTHROPIC_API_KEY): clients[claude-3-5-sonnet] AnthropicClient() return clients def run_analysis(symbol: str, days: int 60): # 1. 准备数据与图表 print(f[1/4] 获取 {symbol} 行情数据...) df fetch_daily_data(symbol, days) chart_path kline.png plot_kline(df, chart_path) print(f[1/4] K线图已生成: {chart_path}) # 2. 初始化模型客户端 print([2/4] 初始化AI模型客户端...) clients build_clients() if not clients: raise RuntimeError(未配置任何API Key请设置 OPENAI_API_KEY 或 ANTHROPIC_API_KEY) # 3. 让每个模型处理每个角色 # 实际项目中可以把每个角色的请求放到线程池中并行执行 print([3/4] 多个AI并行分析中...) results [] for role in MODEL_ROLES: # 这里简单轮询使用客户端生产环境可增加负载均衡 client_name list(clients.keys())[len(results) % len(clients)] client clients[client_name] try: content client.analyze(chart_path, role[prompt]) except Exception as e: content f[错误] {str(e)} results.append({ role_name: role[name], model_name: client_name, content: content, }) print(f - {role[name]} 完成) # 4. 汇总输出 print([4/4] 汇总分析结果...) summary aggregate_results(results) save_report(summary, analysis_report.json) print(f分析报告已保存: analysis_report.json) print( * 40) print(summary[combined_text]) if __name__ __main__: # 示例分析上证指数 run_analysis(symbol000001, days60)4.7 运行与验证在项目根目录执行export OPENAI_API_KEY你的OpenAI Key export ANTHROPIC_API_KEY你的Anthropic Key python main.py预期输出如下内容因模型不同会有差异[1/4] 获取 000001 行情数据... [1/4] K线图已生成: kline.png [2/4] 初始化AI模型客户端... [3/4] 多个AI并行分析中... - 趋势分析 完成 - 形态识别 完成 - 量价分析 完成 [4/4] 汇总分析结果... 分析报告已保存: analysis_report.json 【趋势分析】 当前处于震荡偏多趋势MA5 上穿 MA20 ... 【形态识别】 近期出现十字星形态结合位置看存在变盘可能 ... 【量价分析】 成交量整体持平未见明显放量 ...同时当前目录下会产生kline.png输入给模型的 K 线图。analysis_report.json结构化分析报告。4.8 并行优化上面的主流程是串行调用三个模型实际使用中速度较慢。推荐改为线程池并发请求from concurrent.futures import ThreadPoolExecutor, as_completed def run_parallel_analysis(clients, chart_path): results [] total len(MODEL_ROLES) def _request(role): client_name list(clients.keys())[len(results) % len(clients)] client clients[client_name] try: content client.analyze(chart_path, role[prompt]) except Exception as e: content f[错误] {str(e)} return { role_name: role[name], model_name: client_name, content: content, } with ThreadPoolExecutor(max_workerstotal) as executor: future_map {executor.submit(_request, role): role for role in MODEL_ROLES} for future in as_completed(future_map): results.append(future.result()) return results并行后整体耗时取决于最慢的那个模型调用而不是三个模型耗时的总和能明显提升体验。5. 常见问题与排查思路5.1 模型返回内容为空问题现象常见原因解决思路返回 content 为 None 或空字符串图片解析失败、请求被安全策略拦截检查图片是否为标准PNG尺寸是否过大打印完整响应结构确认返回字段名返回内容包含“抱歉无法分析”提示词不够具体模型未识别为有效任务强化角色设定明确要求分析维度5.2 图片生成后模型识别困难如果模型看不清 K 线图上的信息通常不是模型问题而是图表本身信息密度过高的原因。我们建议把图片宽度控制在 1000 像素左右不要用超大原图。均线数量控制在 2~3 条不要叠加太多指标。中文标注如果识别不好改成英文标注或阿拉伯数字。压缩图片质量到 80% 左右减少 Base64 体积。5.3 不同模型对同一张图结论不一致这是正常现象也是多 AI 分析的价值所在。如果结论完全一致反而说明模型可能受到了共同训练数据的影响。处理策略在汇总报告中明确标注每个模型的“角色”不要求所有模型回答相同问题。对结论做交叉验证如果趋势分析说“上升”量价分析说“缩量上行需谨慎”综合结论应体现这种分歧。不建议用投票方式简单决定最终结论量化交易中需要的是多维度参考而不是民主表决。5.4 API 调用超时多模态请求比纯文本请求慢很多尤其是图片较大的时候问题现象常见原因解决思路请求超过 30 秒无响应图片体积过大、网络不稳定压缩图片、设置更高超时时间部分模型失败影响整体没有做超时隔离使用 as_completed Future.timeout 控制单项等待时间建议在客户端中增加超时参数例如 OpenAI 的timeout60Claude 的timeout120具体数值根据网络情况调整。5.5 成本失控多 AI 分析意味着每次完整分析要调用多次模型成本是单次分析的数倍。控制成本的方法包括控制每次请求的max_tokens不需要模型写长篇分析。对图片做降采样缩小图片体积。设置每日调用上限按用户粒度限流。使用模型缓存层对相同 K 线图不重复调用。6. 最佳实践与工程建议6.1 提示词版本管理AI 量化分析系统的提示词会频繁调整。建议把提示词放入独立的配置文件或数据库中而不是硬编码在代码里。每次修改提示词后记录版本号方便回滚对比。6.2 分析结果的可追溯性每份分析报告必须包含生成时间。使用的模型名称与版本。提示词版本。K 线图数据的时间范围。这样当分析结论出现严重偏差时你可以回到当时的输入条件定位是哪一层出了问题。6.3 熔断与降级生产环境中模型 API 不可用是常态。一定要设计降级策略如果主要模型失败自动切换备用模型。如果所有模型都失败返回“AI分析服务暂不可用”不得返回空报告。监控模型调用成功率连续失败超过阈值时触发告警。6.4 安全边界AI 分析结果只能作为参考不能替代投资决策。在 UI 或报告中应明确标注“AI 生成内容仅供参考不构成投资建议”。这一点既是合规要求也是对用户负责。6.5 缓存策略对于同一标的、同一时间周期的 K 线图在短时间内分析结果差异不大。可以按“标的周期日期”做缓存当天重复请求直接返回缓存结果减少 API 成本和延迟。6.6 日志与监控每个模型的请求都应该记录logger.info({ event: ai_analysis_request, symbol: symbol, model: client_name, role: role_name, cost_tokens: tokens_used, latency_ms: latency_ms, success: success, })这些数据后续可以用于分析模型效果、统计成本、排查故障。7. 总结与下一步学习方向本文实现了一个“多个 AI 同时看一张图并给出各自分析”的完整功能先生成 K 线图再通过多线程方式将图片分别发送给不同角色的大模型最后汇总分析并输出结构化工单。项目核心代码可以在本地直接运行替换 API Key 后即可作为自己的 AI 量化分析工具使用。接下来你可以从这几个方向继续深入接入更多金融数据源例如期货、加密货币、港美股数据。将分析结果接入自己的量化策略回测系统做信号叠加。增加更丰富的图表元素如布林带、MACD但要注意控制信息密度。使用向量数据库存储历史分析报告构建 AI 量化投研知识库。多 AI 分析不是让模型替你决策而是让不同视角的信息帮你减少盲区。真正落地时模型的稳定输出、成本控制和结果回溯往往比“模型本身有多聪明”更重要。希望这篇实战教程能给你带来启发。如果你在实现过程中遇到模型 API 兼容、图表识别或结果汇总方面的问题欢迎留言交流。