Posted by on | Featured

大多数金融人工智能工具都擅长一件事：总结股票的相关信息。当你询问苹果公司、英伟达或特斯拉的情况时，这些工具会为你提供价格走势的概览、一些财务比率数据，或许还会介绍一些公司背景信息。这些信息确实很有用，但一旦任务需要进行更深入的研究时，这些工具就显得力不从心了。

真正的研究通常是从对某个公司的整体分析开始的，而不是仅仅查看它的股票代码。交易员、分析师或产品团队更可能会这样思考：“苹果公司看起来很有投资价值，因为其下行风险已经得到控制，而且业务质量依然很高。但这些数据真的能支持这一观点吗？”这是一个完全不同的问题。简单的总结是无法回答这个问题的，因为系统需要亲自验证这些观点的正确性，而不仅仅是描述该公司的相关情况。

在本次教程中，我们将构建这样一个金融研究辅助工具，它能够完成这样的任务。该工具可以接收用自然语言表述的研究需求，通过EODHD的MCP服务器获取历史价格和财务数据，将这些信息转化为结构化的数据证据，最后生成一份简短的研究报告并给出结论。

目录

先决条件

在开始之前，请确保你已经具备了以下条件。

你需要安装Python 3.9或更高版本，同时还需要以下库：mcp、openai、numpy和pandas。在运行任何代码之前，请使用pip将这些库安装到位。

你还需要两把API密钥：一把来自EODHD，用于获取历史价格和基本数据；另一把来自OpenAI，用于数据解析和备忘录生成。如果你没有EODHD的API密钥，可以在eodhd.com注册开发者账户来获取。

本教程假设读者对Python及异步编程有一定的了解。虽然不需要具备金融领域的专业知识，但在阅读相关内容之前，了解市盈率以及资金回撤率等概念会有所帮助。

建议使用Jupyter笔记本环境来进行各项测试，不过任何支持await的Python环境都可以使用。

Copilot实际生成的内容

在了解整个工作流程之前，先了解一下我们最终会得到什么样的结果会很有帮助。理解这个项目最简单的方法就是看一个具体的示例。

假设用户向系统输入如下指令：

我认为苹果公司很有投资价值，因为其潜在风险已经得到了控制，而且业务质量依然很高。你能帮我检测一下过去180天内AAPL的表现吗？

Copilot并不会简单地对苹果公司进行概述，而是会生成一份结构清晰的研究备忘录：

1. 研究结论  

苹果公司具有较高的投资吸引力，因为其潜在风险得到了有效控制，业务质量也始终保持良好。  

2. 支持性证据  

在过去的180天内，苹果公司的最大资金回撤率仅为-13.82%，这说明其风险相对较低。盈利能力指标也很强劲，营业利润率为35.37%，净利润率为27.04%。资本回报率同样很高，总资产回报率为24.38%，净资产收益率为152.02%，这些数据都表明苹果公司的资产使用效率极高，资本运作也非常高效。从增长指标来看，该公司业务发展势头良好，季度营收同比增长了15.70%，净利润同比增长了18.30%。未来预期也是如此，预计其净利润增长率将为9.68%，营收增长率将为6.87%。  

3. 减弱这一结论的证据  

过去30天内，分析师对苹果公司盈利能力的预期有所下调，净每股收益预期下降了-3%，这说明投资者对苹果公司的看法有所恶化。  

4> 缺失的证据  

提供的数据集中没有发现任何重要的缺失信息。  

5. 最终判断  

部分支持这一结论——支持性证据多于反驳性证据，但这一结论尚未得到完全证实。  

6. 总结性评价  

苹果公司展现了强劲且稳定的业务表现，高利润率、良好的资本回报率以及持续的增长势头都证明了这一点。虽然在观察期内其潜在风险相对较低，但也不能忽视。不过，由于分析师对盈利能力的预期下调，这一结论仍需进一步验证才能确定。

这个例子让该项目的目标变得更加清晰了。我们并不是在构建一个仅仅能告诉我们苹果公司发生了什么变化的系统，而是要构建这样一个系统：它能够接收某种观点或主张，然后将其与市场数据及基本财务信息进行对比分析，最终给出有条理的判断结果。

这种区别非常重要，因为那份备忘录其实只是最终呈现的结果而已。在它的背后，系统首先会解析用户提出的观点或主张，然后通过EODHD的MCP服务器获取相关价格数据及基本财务信息，计算出各种关键指标，分析这些数据对观点的支持作用或矛盾之处，最终得出结论，并据此生成最终的备忘录。正是这样的处理流程才使得输出结果具有条理性和结构性。

在第一步中，我们将构建所有必要的组件，从而形成能够生成这种类型输出结果的系统架构。

这与普通的股票辅助工具有何不同

股票辅助工具与观点分析工具的工作流程对比

普通的股票辅助工具通常会从股票代码开始，试图解释该股票的价格走势。它可能会总结价格变化情况，提及一些财务比率，并提供一些关于该公司的基本信息。当问题比较宽泛时，这样的工具确实很有用；但当输入的是一个具体的投资观点或主张时，这样的工具就远远不够用了。

而这个项目则是从相反的方向着手进行的。它的输入并不是“告诉我关于苹果公司的情况”，而是某种具体的观点或主张，比如“苹果公司看起来很有投资价值，因为其下行风险已经得到控制，而且业务质量依然很高”。这种输入方式改变了系统的工作任务——它现在必须逐一验证这个观点或主张的各个部分，确定哪些因素支持这一观点，哪些因素会削弱它，以及还有哪些信息是缺失的。

正是这种思维方式的转变，决定了整个工作流程的走向。该系统不仅仅局限于数据检索和总结，还必须解析用户提出的观点，将相关数据与适当的证据进行匹配，最终给出明确的结论。正因为如此，这个工具更像是一个研究辅助工具，而不仅仅是一个普通的股票信息汇总工具。

工作流程

从宏观层面来看，这个辅助工具遵循以下简单的流程：

  • 将用户提出的观点解析成结构化请求

  • 通过MCP获取历史价格数据及基本财务信息

  • 将这些数据转化为市场分析指标和业务评估信息

  • 分析这些指标，确定哪些因素支持用户提出的观点，哪些因素会削弱它

  • 最终得出结论

  • 生成最终的备忘录

这就是整个工作流程的完整环节。虽然输出结果看起来像是一份简短的研究报告，但实际上它是建立在core.py中精心设计的系统架构之上的。

项目结构：

project/
├── client.py
├── core.py
└── test.ipynb

client.py是负责与EODHD服务器进行交互的层。它负责连接EODHD服务器，列出可用的工具，以适当的重试次数和超时设置调用这些工具，并为每次请求返回相应的元数据。core.py则包含了实际的观点分析逻辑，包括数据解析、信息获取、指标计算、证据整理、结论生成以及备忘录编写等功能。test.ipynb用于进行质量检测和端到端的测试演示。

这种结构划分非常有用，因为它能让教程的内容更加易于理解。当我们开始编写代码时，每个功能模块都有明确的存放位置：与MCP相关的代码放在client.py文件中，而与研究流程相关的代码则放在core.py文件中。

构建MCP客户端

我们将从项目中最基础的部分开始入手，也就是MCP访问层。

这个文件只负责一项功能：它会连接到EODHD的MCP服务器，列出可用的工具，以一定的重试次数和超时设置来调用这些工具，并在返回响应的同时附带一些元数据信息。实际上，论文的核心逻辑并不应该放在这个文件中。保持这一层的代码量尽可能少，会让后续的开发工作变得更加容易理解。

创建一个名为client.py的文件，然后添加以下代码：

import time
import asyncio

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

class EODHDMCP:
    def __init__(self, apikey, base_url=None):
        self.apikey = apikey
        self.base_url = base_url or "https://mcp.eodhd.dev/mcp"
        self._tools = None

    def _url(self):
        return f"{self.base_url}?apikey={self.apikey}"

    def _open(self):
        return streamablehttp_client(self._url())

    async def list_tools(self):
        if self._tools is not None:
            return self._tools

        async with self._open() as (read, write, _):
            async with ClientSession(read, write) as s:
                await s.initialize()
                resp = await s.list_tools()
                self._tools = [t.name for t in resp.tools]
                return self._tools

    async def call_tool(self, name, args, trace_id, timeout_s=25, retries=2):
        last = None

        for attempt in range(retries + 1):
            t0 = time.time()
            try:
                async with self._open() as (read, write, _):
                    async with ClientSession(read, write) as s:
                        await s.initialize()
                        out = await asyncio.wait_for(s.call_tool(name, args), timeout=timeout_s)
                        dt = time.time() - t0
                        meta = {
                            "trace_id": trace_id,
                            "tool": name,
                            "args": args,
                            "latency_s": round(dt, 3),
                        }
                        return out, meta
            except Exception as e:
                last = e
                if attempt < retries:
                    await asyncio.sleep(0.5 * (attempt + 1))

        raise last

在这里，真正重要的只有两个方法。list_tools()这个方法主要用于查看并缓存MCP服务器提供的工具信息；而call_tool()才是项目中其他部分会实际使用的方法。它负责发送请求、处理超时和重试机制，并同时返回原始输出结果以及元数据信息。

这些元数据在后续使用中会非常有用，因为工作流程的轨迹是可以被追踪的。当辅助工具返回结果时，我们仍然能够知道使用了哪种工具、传入了哪些参数，以及整个过程花费了多长时间。因此，尽管这个文件体积很小，但它为系统的其他部分提供了一个清晰、便于检测的数据访问层。

配置core.py

现在MCP客户端已经准备就绪，我们可以开始在core.py中构建主要的工作流程了。

这个文件将包含实际的论文测试逻辑，因此第一步就是设置导入语句、API客户端、一些限制条件，以及一些可供后续流程重复使用的辅助函数。

创建一个名为core.py的文件，首先编写以下内容：

import json
import re
import time
import uuid
import asyncio
from datetime import date, timedelta

import numpy as np
import pandas as pd
from openai import OpenAI

from client import EODHDMCP

eodhd_api_key = "你的EODHD API密钥"
mcp_base_url = "https://mcp.eodhd.dev/mcp"

openai_api_key = "你的OpenAI API密钥"
model_name = "gpt-5.3-chat-latest"

max_lookback_days = 365
max_tool_calls = 10
max_tickers = 5

mcp = EODHDMCP(eodhd_api_key, base_url=mcp_base_url)
oa = OpenAI(api_key=openai_api_key)

def log_event(event, trace_id, **extra):
    payload = {
        "event": event,
        "trace_id": trace_id,
        "ts": round(time.time(), 3),
    }
    payload.update(extra)
    print(json.dumps(payload, default=str))

def get_dates_from_lookback(days):
    end = date.today()
    start = end - timedelta(days=int(days))
    return start.isoformat(), end isoformat()

def make_state():
    return {
        "tool_calls": 0,
        "tool_trace": [],
    }

def bump_tool_call(state, meta):
    state["tool_calls"] += 1
    state["tool_trace"].append.meta)

    if state["tool_calls"] > max_tool_calls:
        raise RuntimeError("工具调用次数超出限制")

def to_text(out):
    if isinstance(out, str):
        return out.strip()

    if hasattr(out, "content"):
        try:
            parts = []
            for item in out.content:
                if hasattr(item, "text") and item.text is not None:
                    parts.append(item.text)
                else:
                    parts.append(str(item))
            return "\n".join(parts).strip()
        except Exception:
            pass

    return str(out).strip()

注意：请将“你的EODHD API密钥”替换为你的实际EODHD API密钥。如果你还没有这个密钥，可以通过注册一个EODHD开发者账户来获取它。

这段代码完成了三件事：

  • 首先，它配置了我们需要使用的两个客户端。mcp是来自client.py的EODHD MCP客户端，而oa则是用于后续解析和生成备忘录的OpenAI客户端。

  • 其次，它为工作流程设置了一些限制条件。这些限制有助于控制系统的运行范围，比如限定回溯窗口的大小、可处理的股票代码数量，以及单次运行中允许进行的工具调用次数。

  • 第三，它添加了一些辅助函数，这些函数被文件中的其他部分所依赖。log_event()用于实现轻量级的追踪功能，get_dates_from_lookback()可以将回溯窗口转换为起始日期和结束日期，make_state()和bump_tool_call()有助于监控MCP的使用情况，而to_text()则可以在我们解析数据之前，将工具的输出结果安全地转换成纯文本形式。

将研究提示转换为结构化请求

这个辅助系统首先需要对输入内容进行整理。用户每次提交的请求都不会具备完美的格式；他们更有可能用普通的英语写下自己的研究思路，并将研究对象、时间范围等信息混杂在同一个提示中。

因此，系统会先将原始的提示信息分解为四个字段：

  • 研究对象

  • 回顾时间范围

  • 研究主题

  • 操作模式

这种处理逻辑被编写在core.py文件中。

def parse_request(text):
    prompt = f"""
您正在为这个金融研究辅助系统提取所需字段。

请仅返回格式正确的JSON数据，其结构应为：
{{
  "tickers": ["AAPL"],
  "lookback_days": 180,
  "thesis": "具体研究主题",
  "mode": "single"
}}

规则如下：
- 仅提取明确提及或可推断出的研究对象。
- 不允许随意创造新的研究对象代码。
- 如果包含多个研究对象，操作模式必须为“watchlist”；
- 如果只包含一个研究对象，操作模式必须是“single”。
- 如果没有指定时间范围，默认使用180天。
- 将月份转换为天数时，按每月30天计算；将年份转换为天数时，按每年365天计算。
- 研究主题应简洁明了，同时准确反映用户的意图。
- 仅返回JSON格式的数据，不允许使用markdown或添加解释。

用户输入：
{text}
"""'.strip()

    r = oaresponses.create(
        model=model_name,
        input=[{"role": "user", "content": prompt}],
    )

    raw = r.output_text.strip()

    try:
        parsed = json.loads(raw)
    except Exception:
        raise RuntimeError(f"解析器返回了非JSON格式的数据：{raw[:500]}")

    return parsed

这个函数为模型指定了非常具体的任务——它并不要求模型进行任何分析或判断，而只是要求它从输入数据中提取出结构化信息。这一点非常重要，因为我们在输入层需要灵活性，但不希望整个处理流程变得混乱不清。

一旦模型返回了JSON格式的数据，Python程序就会对这些数据进行进一步整理和优化。

def enforce_limits(parsed):
    tickers = parsed.get("tickers", [])
    if not isinstance(tickers, list):
        tickers = []

    tickers = [str(x).upper().strip() for x in tickers if str(x).strip()]
    tickers = tickers[:max_tickers]

    lookback_days = parsed.get("lookback_days", 180)
    try:
        lookback_days = int(lookback_days)
    except Exception:
        lookback_days = 180

    if lookback_days < 1:
        lookback_days = 1
    if lookback_days > max_lookback_days:
        lookback_days = max_lookback_days

    thesis = str(parsed.get("thesis", "")).strip()
    if not thesis:
        thesis = "未提供研究主题。"

    mode = parsed.get("mode", "single")
    if len(tickers) > 1:
        mode = "watchlist"
    else:
        mode = "single"

    return {
        "tickers": tickers,
        "lookback_days": lookback_days,
        "thesis": thesis,
        "mode": mode,
    }

这个第二个功能正是维持工作流程正常运转的关键。它负责清理相关数据，限制每次请求中允许获取的数据数量，控制时间窗口的范围，并确保所使用的模式与实际需要处理的数据量相匹配。这样一来，模型为我们提供了灵活性，而代码则为我们的操作设定了界限。这种结合对于构建这样的系统来说非常重要。

获取两种数据来源：历史数据与基本面数据

一旦请求被解析完毕，下一步就是提取那些将用于后续工作流程的数据。在这个版本中，我们仅使用EODHD提供的两个数据来源：历史价格数据和基本面数据。这些数据已经足够用来测试多种类型的分析主题了，而且不会使整个系统的规模变得过于庞大。

请将以下两个功能添加到core.py文件中：

async def fetch_prices(ticker, start_date, end_date, trace_id, state):
    args = {
        "ticker": ticker,
        "start_date": start_date,
        "end_date": end_date,
        "period": "d",
        "order": "a",
        "fmt": "json",
    }

    out, meta = await mcp.call_tool("get_historical_stock_prices", args, trace_id)
    text = to_text(out)

    bump_tool_call(state, meta)

    if not text:
        raise RuntimeError("从get_historical_stock_prices函数获得的响应为空")

    try:
        data = json.loads(text)
    except Exception:
        raise RuntimeError("价格数据解析失败，返回的内容不是JSON格式：{text[:300]}")

    if isinstance(data, dict) and data.get("error"):
        raise RuntimeError(data["error"])

    df = pd.DataFrame(data)
    if df.empty:
        return df

    keep = [c for c in ["date", "close"] if c in df.columns]
    df = df[keep].copy()
    df["ticker"] = ticker

    return df

async def fetch_fundamentals(ticker, trace_id, state):
    args = {
        "ticker": ticker,
        "include_financials": False,
        "fmt": "json",
    }

    out, meta = await mcp.call_tool("get_fundamentals_data", args, trace_id)
    text = to_text(out)

    bump_tool_call(state, meta)

    if not text:
        raise RuntimeError("从get_fundamentals_data函数获得的响应为空")

    try:
        data = json.loads(text)
    except Exception:
        raise RuntimeError("基本面数据解析失败，返回的内容不是JSON格式：{text[:300]}")

    if isinstance(data, dict) and data.get("error"):
        raise RuntimeError(data["error"])

    return data

  • fetch_prices()函数会获取请求指定时间窗口内的每日历史数据，并将其简化为我们现在真正需要的字段：date、close以及股票代码本身。最终得到的这个精简后的DataFrame将被用于后续的计算，比如计算跌幅、波动率、趋势等市场指标。

  • fetch_fundamentals()函数将获取到的基本面数据以JSON格式保存下来，因为我们在后续的部分会从中提取不同的信息，包括利润率、增长情况、估值水平、数据修订内容以及贝塔系数等等。

这里有几点需要注意。这两个函数都是通过同一个MCP封装层来运行的，因此它们会自动继承我们在client.py中实现的所有超时处理、重试机制以及元数据管理功能。此外，这两个函数都会调用bump_tool_call()这个函数，这使我们能够追踪在单次运行过程中究竟进行了多少次外部调用。当以后我们需要让整个工作流程保持可追溯性、而不是成为一个“黑箱”时，这一功能就会派上用场。

利用价格数据构建第一层证据体系

一旦获取到了价格数据，下一步就是将这些原始数据转化为我们可以用来进行分析的形式。对于这个辅助分析工具来说，价格历史数据虽然不是最最终的结论，但仍然构成了第一层证据体系。这些数据有助于我们验证关于风险控制、波动性、市场趋势以及投资回报质量等方面的假设。

将以下代码添加到core.py文件中：

def compute_pricesignals(prices_df):
    if prices_df is None or prices_df.empty:
        return {}

    df = prices_df.copy()
    df["date"] = pd.to_datetime(df["date"], errors="coerce")
    df["close"] = pd.to_numeric(df["close"], errors="coerce")

    df = df.dropna(subset=["date", "close")).sort_values("date")
    if df.empty:
        return {}

    close = df["close"]
    rets = close.pct_change().dropna()

    out = {
        "n_points": int(len(close)),
        "start_price": float/close.iloc[0]),
        "end_price": float(close.iloc[-1]),
    }

    if len(close) >= 2:
        out["ret_total"] = float(close.iloc[-1] / close.iloc[0] - 1)

    if not rets.empty:
        vol_daily = float(rets.std())
        vol_annualized = float(vol_daily * np.sqrt(252))

        out["vol_daily"] = vol_daily
        out["vol_annualized"] = vol_annualized

        if vol_annualized > 0 and "ret_total" in out:
            out["ret_to_vol"] = float(out["ret_total"] / vol_annualized)

    peak = close.cummax()
    drawdown = close / peak - 1
    out["max_drawdown"] = float(drawdown.min())

    logp = np.log(close.values)
    x = np.arange(len(logp))
    if len(logp) >= 3:
        out["trend_slope"] = float(np.polyfit(x, logp, 1)[0])
    else:
        out["trendslope"] = 0.0

    return out

这个函数能够从单纯的价格数据中提取出一组有用的市场信号。ret_total反映了股票在整个观察周期内的走势变化：vol_annualized则说明了这种走势的波动程度；当讨论风险控制问题时，max_drawdown这个指标非常有用；trend_slope可以用来判断市场趋势的方向，而ret_to_vol则有助于我们更全面地评估投资回报的质量，而不仅仅关注原始的回报数值。

关键在于，我们并没有让模型直接从原始价格数据中推导出所有这些信息。这些计算都是先在Python中完成的，因此后续的分析过程是基于明确的信号进行的，而不是基于模糊的解读。这样的设计使得整个工作流程更加稳定可靠。

从基本面数据构建第二层证据体系

价格数据为我们提供了论证的一个方面，而另一个方面则来自基本面分析。正是这部分内容使得分析结果不再显得千篇一律、缺乏针对性。一旦辅助系统开始将基本面数据视为真正的证据，而不仅仅是公司的基本概况信息，那么分析结果就会变得有用得多。

首先在core.py中添加这个辅助函数：

def _to_float(x):
    if x in (None, "", "NA"):
        return None
    try:
        return float(x)
    except Exception:
        return None

这个简单的函数用于在我们使用数据之前对它们进行清洗。基本面分析中包含的资料往往包含字符串、空值或"NA"，因此提前对这些数据进行标准化处理是非常有必要的。

现在再添加主函数：

def compute_fundamentalsignals(fundamentals):
    if not isinstance(fundamentals, dict):
        return {}

    general = fundamentals.get("General", {}) or {}
    highlights = fundamentals.get("Highlights", "") or {}
    valuation = fundamentals.get("Valuation", "") or {}
    technicals = fundamentals.get("Technicals",\") or {}

    earnings = fundamentals.get("Earnings", "") or {}
    trend = earnings.get("Trend", "") or {}

    latest_trend = None
    if isinstance(trend, dict) and trend:
        latest_key = sorted(trend.keys())[-1]
        latest_trend = trend.get(latest_key, {}) or {}
    else:
        latest_trend = {}

    out = {
        "sector": general.get("Sector"),
        "industry": general.get("Industry"),
        "employees": _to_float(general.get("FullTimeEmployees")),

        "market_cap": _to_float(highlights.get("MarketCapitalization")),
        "pe_ratio": _to_float(highlights.get("PERatio')),
        "peg_ratio": _to_float(highlights.get("PEGRatio"));
        "profit_margin": _to_float(highlights.get("ProfitMargin"));
        "operating_margin": _to_float(highlights.get("OperatingMarginTTM"));
        "roa": _to_float(highlights.get("ReturnOnAssetsTTM"));
        "roe": _to_float(highlights.get("ReturnOnEquityTTM"));
        "revenue_ttm": _to_float(highlights.get("RevenueTTM"));
        "revenue_growth_yoy": _to_float(highlights.get("QuarterlyRevenueGrowthYOY"));
        "earnings_growth_yoy": _to_float(highlights.get("QuarterlyEarningsGrowthYOY"));
        "dividend_yield": _to_float(highlights.get("DividendYield")),

        "trailing_pe": _to_float(valuation.get("TrailingPE"));
        "forward_pe": _to_float(valuation.get("ForwardPE"));
        "price_sales": _to_float(valuation.get("PriceSalesTTM"));
        "price_book": _to_float(valuation.get("PriceBookMRQ"));
        "ev_revenue": _to_float(valuation.get("EnterpriseValueRevenue"));
        "ev_ebitda": _to_float(valuation.get("EnterpriseValueEbitda")),

        "beta": _to_float(technicals.get("Beta"));

        "earnings_estimate_growth": _to_float(latest_trend.get("earningsEstimateGrowth"));
        "revenue Estimate_growth": _to_float(latest_trend.get("revenueEstimateGrowth"));
        "eps_revisions_up_30d": _to_float(latest_trend.get("epsRevisionsUpLast30days"));
        "eps_revisions_down_30d": _to_float(latest_trend.get("epsRevisionsDownLast30days"));
    }

    if out["trailing_pe"] is not None and out["forward_pe"] is not None:
        out["forward_vs_trailing_pe_change"] = out["forward_pe"] - out["trailing_pe"]

    if out["eps_revisions_up_30d"] is not None and out["eps_revisions_down_30d"] is not None:
        out["net_eps_revisions_30d"] = out["eps_revisions_up_30d"] - out["eps_revisions_down_30d"]

    return out

这个功能汇集了那些对论文分析最为重要的基本数据要素。

  • 从Highlights部分，我们可以获取盈利能力、资本回报率、增长速度以及市值等信息。而从Valuation部分，我们能得到市盈率、市销率以及基于企业价值的各类比率。

  • 从Technicals部分，我们可以获取股票的贝塔系数。

  • 从Earnings.Trend部分，我们可以获得对未来公司增长的预测数据以及这些预测数据的调整情况。

这些数据使我们能够以更加具体的方式来检验有关企业质量、估值合理性以及未来发展预期的各种观点。

最后两个衍生出来的数据字段也同样很有用。市盈率与过去一段时间的市盈率的差值，可以帮助我们快速判断公司的估值状况是在趋于合理还是仍然偏高；而过去30天内分析师对每股净利润预测的调整情况，则能说明他们的预期是在改善还是恶化。

目前我们已经掌握了哪些信息？

目前，这个辅助系统已经能够解析论文分析的内容，获取相关价格数据和基本财务数据，并将这些数据转化为两种可供后续分析使用的信号类型：

  • 价格相关的信号包括回报率、波动性、回撤幅度、趋势以及回报的质量等。

  • 基本财务数据相关的信号包括利润率、资本回报率、增长速度、估值水平、各项数据的调整情况以及贝塔系数等。

接下来，我们会将这些信号转化为真正的研究工作流程所需要的形式：包括支持性证据、削弱论点的依据、遗漏的信息、最终结论以及最终的报告摘要。

对论文分析内容进行分类

在辅助系统能够对某篇论文分析做出判断之前，它首先需要了解这篇分析所针对的具体观点是什么。

这一点非常重要，因为并不是所有的论文分析都应该采用相同的评估方法。对于那些涉及“控制下行风险”这一主题的分析来说，回撤幅度和波动性应该是更重要的考量因素；而对于那些关注企业质量的分析而言，利润率、资本回报率和增长速度则更为关键；而那些涉及到估值合理性的分析，则可能需要同时考虑企业质量和估值这两个方面。

因此，在直接根据各种数据得出结论之前，我们会先进行一个简单的分类步骤。这样系统就能明确知道需要分析哪些类型的观点，从而更清晰地理解整篇论文的分析内容。

将以下代码添加到core.py文件中：

def classify_thesis(thesis):
    prompt = f"""
您正在将这篇关于某只股票的论文分析分类为几种主要的观点类型。

请仅返回如下格式的有效JSON数据：
{{
  "claim_types": ["controlled_downside", "business_quality"],
  "summary": "对这篇论文分析的简短总结"
}}

允许的观点类型包括：
- controlled_downside
- momentum_strength
- low_risk
- high_risk
- valuation_attractive
- valuation_expensive
- business_quality
- weak_business_quality
- premium_justified
- premium_not_justified

规则如下：
- 仅选择与分析内容明显相关的观点类型
- 不要随意创建新的分类标签
- 如果没有合适的分类选项，返回空列表
- 总结部分应简洁且准确反映论文分析的核心内容

论文分析内容：
{thesis}
"".strip()

    r = oaresponses.create(
        model=model_name,
        input=[{"role": "user", "content": prompt}],
    )

    raw = r.output_text.strip()

    try:
        out = json.loads(raw)
    except Exception:
        raise RuntimeError(f"论文分析分类工具返回了非JSON格式的数据：{raw[:500]}")

    claim_types = out.get("claim_types", [])
    if not isinstance(claim_types, list):
        claim_types = []

    clean = []
    allowed = {
        "controlled_downside",
        "momentum_strength",
        "low_risk",
        "high_risk",
        "valuation_attractive",
        "valuation_expensive",
        "business_quality",
        "weak_business_quality",
        "premium_justified",
        "premium_not_justified",
    }

    for x in claim_types:
        x = str(x).strip()
        if x in allowed and x not in clean:
            clean.append(x)

    return {
        "claim_types": clean,
        "summary": str(out.get("summary", "")).strip(),
    }

这个功能使模型的工作范围更加明确。它并不需要判断某个论点是对还是错，而只需要确定该论点属于哪一类。这样一来，后续的处理步骤就会变得更加简洁，因为证据分析系统就不必再以相同的方式处理所有的输入信息了。

最后的验证步骤也同样重要。尽管模型会返回各种标签，但Python仍会通过预定义的规则对这些标签进行筛选，剔除任何不符合要求的因素。这样就能保证这个环节既具有灵活性，又受到有效的控制。

将信号转化为支持论据、矛盾点及缺失证据

在这个阶段，辅助系统才开始真正进行推理分析。

到目前为止，我们已经掌握了三样关键信息：论点本身、各种论点类型，以及由价格数据和基本面因素构建出来的信号数据。然而，除非这些信息能够被转化为明确的论证依据，否则它们本身并无实际意义。

这意味着对于每一个论点来说，系统都需要回答三个问题：

  • 数据中有哪些因素能够支持这一论点？

  • 数据中又有哪些因素会削弱这一论点？

  • 要准确判断这一论点的合理性，还缺少哪些信息？

而`build_evidence_blocks()`函数正是用来完成这些工作的。它会对已分类的论点进行分析，检查相关的价格信号和基本面数据，并将这些信息分为三类：支持论据、矛盾点以及缺失证据。

请将以下代码添加到`core.py`文件中：

def build_evidence_blocks(thesis, thesis_tags, pricesignals, fundamental_signals):
    evidence_for = []
    evidence_against = []
    missing_evidence = []

    ret_total = price Signals.get("ret_total")
    vol = priceSignals.get("vol_annualized")
    dd = priceSignals.get("max_drawdown")
    trend = priceSignals.get("trend_slope")
    ret_to_vol = priceSignals.get("ret_to_vol")

    pe = fundamentalsignals.get("pe_ratio") or fundamental Signals.get("trailing_pe")
    forward_pe = fundamentalsignals.get("forward_pe")
    beta = fundamentalsignals.get("beta")

    profit_margin = fundamentalsignals.get("profit_margin")
    operating_margin = fundamentalsignals.get("operating_margin")
    roa = fundamentalsignals.get("roa")
    roe = fundamentalsignals.get("roe")
    revenue_growth = fundamentalsignals.get("revenue_growth_yoy")
    earnings_growth = fundamentalsignals.get("earnings_growth_yoy")
    earnings_estimate_growth = fundamentalsignals.get("earnings_estimate_growth")
    revenue_estimate_growth = fundamentalsignals.get("revenue_estimate_growth")
    net_eps_revisions = fundamentalsignals.get("net.eps_revisions_30d")

    claim_types = thesis_tags.get("claim_types", "")

    if "controlled_downside" in claim_types:
        if dd is not None:
            if dd > -0.15:
                evidence_for.append(f“最大回撤幅度控制在{dd:.2%}以内。”)
            else:
                evidence_against.append(f“最大回撤幅度达到了{dd:.2%}, 这削弱了‘控制下行风险’这一论点。”)
        else:
            missing_evidence.append("没有可用的回撤数据来验证下行风险控制情况。")

    if "momentum_strength" in claim_types:
        if trend is not None and ret_total is not None:
            if trend > 0 and ret_total > 0:
                evidence_for.append(f“市场走势呈上升趋势，且总回报率为{ret_total:.2%}。”)
            else:
                evidence_against.append("市场走势和总回报率无法有效支持‘动能强劲’这一论点。”)
        else:
            missing_evidence.append("没有可用的趋势或回报数据来验证动能情况。")

    if "low_risk" in claim_types:
        if vol is not None:
            if vol < 0.30:
                evidence_for.append(f“年化波动率为{vol:.2%}, 这支持‘低风险’这一论点。”)
            else:
                evidence_against.append(f“年化波动率为{vol:.2%}, 这削弱了‘低风险’这一论点。”)
        else:
            missing_evidence.append("没有可用的波动率数据来验证风险情况。」)

    if "high_risk" in claim_types:
        if vol is not None:
            if vol >= 0.30:
                evidence_for.append(f“年化波动率为{vol:.2%}, 这支持‘高风险’这一论点。”)
            else:
                evidence_against.append(f“年化波动率为{vol:.2%}, 这并不能有效支持‘高风险’这一论点。”)
        else:
            missing_evidence.append("没有可用的波动率数据来验证风险情况。」)

    if "valuation_attractive" in claim_types:
        if pe is not None:
            if pe < 20:
                evidence_for.append(f“市盈率为{pe:.2f}, 这支持‘估值具有吸引力’这一论点。”)
            elif pe > 30:
                evidence_against.append(f“市盈率为{pe:.2f}, 这削弱了‘估值具有吸引力’这一论点。”)
        else:
            missing_evidence.append("没有可用的市盈率数据来验证估值情况。」)

        if forward_pe is not None and pe is not None:
            if forward_pe < pe:
                evidence_for.append(f“未来市盈率为{forward_pe:.2f}, 低于当前的市盈率{pe:.2f}，这可以说明公司的盈利状况正在改善。”)
            else:
                evidence_for.append("未来市盈率和当前市盈率相同，无法用来支持或削弱这一论点。")

        if "valuation_expensive" in claim_types or "premium_not_justified" in claim_types:
            if pe is not None:
                if pe > 30:
                    evidence_for.append(f“市盈率为{pe:.2f}, 这支持‘估值过高’这一论点。”)
            else:
                evidence_against.append(f“市盈率为{pe:.2f}, 这并不能有效支持或削弱‘估值过高’这一论点。”)
        else:
            missing_evidence.append("没有可用的市盈率数据来验证估值情况。」)

    if "business_quality" in claim_types or "premium_justified" in claim_types:
        quality_hits = 0

        if operating_margin is not None:
            if operating_margin >= 0.25:
                evidence_for.append(f“营业利润率为{operating_margin:.2%}, 这说明企业的经营质量较高。”)
                qualityhits += 1
            else:
                evidence_against.append(f“营业利润率为{operating_margin:.2%}, 这并不符合‘经营质量高’这一论点的要求。”)
        else:
            missing_evidence.append("没有可用的营业利润率数据来验证经营质量情况。」)

        if profit_margin is not None:
            if profit_margin >= 0.20:
                evidence_for.append(f“利润率为{profit_margin:.2%}, 这说明企业的盈利能力较强。”)
                quality_hits += 1
            else:
                evidence_against.append(f“利润率为{profit_margin:.2%}, 这削弱了‘盈利能力强’这一论点。”)
        else:
            missing_evidence.append("没有可用的利润率数据来验证盈利能力情况。」)

        if roa is not None:
            if roa >= 0.10:
                evidence_for.append(f“资产回报率率为{roa:.2%}, 这说明企业的资产利用效率较高。”)
                quality_hits += 1
            else:
                evidence_against.append(f“资产回报率率为{roa:.2%}, 这并不能有效支持‘资产利用率高’这一论点。”)
        else:
            missing_evidence.append("没有可用的资产回报率数据来验证资产利用效率情况。」)

        if roe is not None:
            if roe >= 0.20:
                evidence_for.append(f“股本回报率为{roe:.2%}, 这说明企业的资本运用效率较高。”)
                quality_hits += 1
            else:
                evidence_against.append(f“股本回报率为{roe:.2%}, 这与‘资本运用效率高’这一论点的要求不符。”)
        else:
            missing_evidence.append("没有可用的股本回报率数据来验证资本运用效率情况。」)

        if revenue_growth is not None:
            if revenue_growth > 0:
                evidence_for.append(f“季度营收增长率为{revenue_growth:.2%}，这支持‘企业发展势头强劲’这一论点。”)
                quality_hits += 1
            else:
                evidence_against.append(f“季度营收增长率为{revenue_growth:.2%}，这削弱了‘企业发展势头强劲’这一论点。”)
        else:
            missing_evidence.append("没有可用的营收增长率数据来验证企业发展势头情况。」)

        if earnings_growth is not None:
            if earnings_growth > 0:
                evidence_for.append(f“季度盈利增长率为{earnings_growth:.2%}，这支持‘企业盈利能力强劲’这一论点。”)
                quality_hits += 1
            else:
                evidence_against.append(f“季度盈利增长率为{earnings_growth:.2%}，这削弱了‘企业盈利能力强劲’这一论点。”)
        else:
            missing_evidence.append("没有可用的盈利增长率数据来验证企业盈利能力情况。」)

        if earnings_estimate_growth is not None:
            if earnings Estimate_growth > 0:
                evidence_for.append(f“未来盈利增长率预计为{earnings_estimate_growth:.2%}, 这支持‘企业未来发展前景良好’这一论点。”)
            else:
                evidence_against.append(f“未来盈利增长率预计为{earnings_estimate_growth:.2%}, 这削弱了‘企业未来发展前景良好’这一论点。”)
        else:
            missing_evidence.append("没有可用的未来盈利增长率数据来验证企业未来发展前景情况。」)

        if revenue Estimate_growth is not None:
            if revenue_estimate_growth > 0:
                evidence_for.append(f“未来营收增长率预计为{revenue_estimate_growth:.2%}, 这支持‘企业未来发展势头强劲’这一论点。”)
            else:
                evidence_against.append(f“未来营收增长率预计为{revenue_estimate_growth:.2%}, 这削弱了‘企业未来发展势头强劲’这一论点。”)
        else:
            missing_evidence.append("没有可用的未来营收增长率数据来验证企业未来发展前景情况。」)

        if net_eps_revisions is not None:
            if net.eps_revisions > 0:
                evidence_for.append(f“过去30天内，每股净利润预期有所增加，增幅为{net_eps_revisions:.0f}，这支持‘企业未来预期良好’这一论点。”)
            else:
                evidence_against.append(f“过去30天内，每股净利润预期有所减少，降幅为{net.eps_revisions:.0f}，这削弱了‘企业未来预期良好’这一论点。”)
        else:
            missing_evidence.append("没有可用的每股净利润预期变化数据来验证企业未来预期情况。」)

        if quality_hits == 0:
            missing_evidence.append("当前数据不足以有效支持或反驳‘企业质量高’这一论点。」)

    if "weak_business_quality" in claim_types:
        if operating_margin is not None and operating_margin < 0.15:
            evidence_for.append(f“营业利润率为{operating_margin:.2%}, 这说明企业的经营质量较低。”)
        if profit_margin is not None and profit_margin < 0.10:
            evidence_for.append(f“利润率为{profit_margin:.2%}, 这说明企业的盈利能力较弱。”)
        if revenue_growth is not None and revenue_growth <= 0:
            evidence_for.append(f“季度营收增长率为{revenue_growth:.2%}，这说明企业的发展势头不佳。”)
        if earnings_growth is not None and earnings_growth <= 0:
            evidence_for.append(f“季度盈利增长率为{earnings_growth:.2%}，这说明企业盈利能力较弱。”)
        else:
            missing_evidence.append("没有可用的数据来验证企业经营质量较低的情况。」)

    if beta is not None:
        if beta > 1.2:
            evidence_against.append(f“贝塔值为{beta:.2f}, 这表明该股票的敏感度高于市场平均水平。”)
        elif beta < 0.9:
            evidence_for.append(f“贝塔值为{beta:.2f}, 这表明该股票的敏感度低于市场平均水平。”)
    else:
        missing_evidence.append("没有可用的贝塔值数据来验证股票的市场敏感度情况。」)

    if ret_to_vol is None:
        missing_evidence.append("没有可用的回报率与波动率之间的关系数据来验证相关论点。)"

    if not evidence_for and not evidence_against:
        missing_evidence.append("当前的数据不足以有效支持或反驳这一论点。”)

    return {
        "thesis": thesis,
        "thesis_summary": thesis_tags.get("summary", ""),
        "claim_types": claim_types,
        "evidence_for": evidence_for,
        "evidence_against": evidence_against,
        "missing_evidence": list(dict.fromkeys(missing_evidence)),
    }

这个函数看起来很长，但一旦你将其逻辑分解开来，就会发现其实很简单。

它首先从我们之前构建的这两个证据层中提取所需的信号，然后逐一检查这些论点标签。如果论点是关于“控制下行风险”的，那么它会关注资产回撤幅度；如果是关于风险的，就会考察波动性及贝塔系数；如果是关于企业质量的，则会参考利润率、资本回报率、成长性以及相关数据修订情况；而如果是关于估值的，就会检查市盈率等指标，以及未来估值与过去估值之间的关系。

这就是这个项目中的关键所在：辅助系统不再仅仅是收集数据，而是会判断在EODHD支持的信号集中，哪些部分实际上与当前正在分析的论点相关。

这三个输出结果使得这个系统真正具备了实用性。

  • evidence_for用于存储支持该论点的证据。
  • evidence_against用于存储削弱该论点的证据。
  • missing_evidence用于明确指出那些缺失的证据，避免系统给出过于自信的结论。

正因如此，这个流程更像是一个用于验证论点的分析流程，而不仅仅是一份经过整理的股票分析报告。

合理性检查（Jupyter Notebook）

你可以在test.ipynb文件中运行这段代码来进行快速合理性检查：

import uuid
from core import (
    fetch_prices,
    fetch_fundamentals,
    compute_pricesignals,
    classify_thesis,
    build_evidence_blocks,
    make_state
)
import json

trace_id = uuid.uuid4().hex[:10]
state = make_state()

thesis = "苹果公司看起来很有投资吸引力，因为其下行风险得到了有效控制，同时企业质量也依然很高。"

prices = await fetch_prices("AAPL.US", "2026-01-01", "2026-04-01", trace_id, state)
funds = await fetch_fundamentals("AAPL.US", trace_id, state)

signals = compute_price_signals(prices)
tags = classify_thesis(thesis)
evidence = build_evidence_blocks(thesis, tags, signals, funds)

print(tags)
print(json.dumps(evidence, indent=2))

预期输出：
合理性检查的预期输出

给出最终判断

一旦这些证据被整理好，辅助系统在生成报告之前还需要再进行一步处理：它需要以一种系统化的方式为这个论点确定一个分类标签。

decide_verdict()函数就是负责完成这一工作的。它会分析有多少证据支持该论点，又有多少证据削弱了它，同时还会判断该论点是否还依赖于某些缺失的企业质量或估值相关数据。这里的目标并不是创建一个完美的评分模型，而是确保系统不会仅仅根据少量的证据就得出过于自信的结论。

请将这段代码添加到core.py文件中：

def decide_verdict(evidence, claim_types=None):
claim_types = claim_types or []

evidence_for = evidence.get("evidence_for", [])
evidence_against = evidence.get("evidence_against",[])
missing = evidence.get("missing_evidence", "")

n_for = len(evidence_for)
n_against = len(evidence_against)
n_missing = len(missing)

quality_claim = any(x in claim_types for x in ["business_quality", "weak_business_quality", "premium_justified", "premium_not_justified"])
valuationClaim = any(x in claim_types for x in ["valuation_attractive", "valuation_expensive", "premium_justified", "premium_not_justified"])

if n_for == 0 and n_against == 0:
return {
"verdict": "unresolved_due_to_missing_evidence",
"reason": "没有足够的可用证据来验证该论点。",
}

if quality_claim and n_missing >= 1:
if n_against > 0:
return {
"verdict": "weakly_supported",
"reason": "部分证据支持该论点，但缺乏直接反映业务质量的证据，且存在矛盾的信息。",
}
return {
"verdict": "partiallysupported",
"reason": "部分论点得到了支持，但缺乏直接反映业务质量的证据。",
}

if valuation_claim and n_missing >= 1:
return {
"verdict": "unresolved_due_tomissing_evidence",
"reason": "该论点所依赖的评估证据在当前数据中并不存在。",
}

if n_for > 0 and n_against == 0:
if n_missing >= 2:
return {
"verdict": "partially_supported",
"reason": "现有的证据支持该论点，但仍有重要证据缺失。",
}
return {
"verdict": "supported",
"reason": "现有证据能够充分支持该论点。",
}

if n_against > 0 and n_for == 0:
return {
"verdict": "not_supported",
"reason": "现有证据主要削弱了该论点的合理性。",
}

if n_for > n_against:
return {
"verdict": "partiallysupported",
"reason": "支持该论点的证据多于反对证据，但论点并未得到完全证实。",
}

if n_against >= n_for:
return {
"verdict": "weakly_supported",
"reason": "反对证据的影响较为显著，因此该论点只能得到弱支持。",
}

return {
"verdict": "unresolved_due_to_missing_evidence",
"reason": "现有证据存在矛盾，无法明确判断该论点的真假。",
}

这里的逻辑设计得相当简单，并不试图进行细致的评分。而是通过分析证据的类型和数量来判断该论点是得到支持、部分支持、弱支持、未得到支持，还是仍然无法确定。

有那么几项检查比其他所有检查都更为重要。如果某个论点依赖于具有商业价值或评估意义的证据，而这些证据却尚未被提供，那么最终的判断结果往往会过早地得出结论，而不会显得那么有说服力。这一点非常重要，因为仅从价格表现来看，某个论点可能看起来很有道理，但如果其所依据的基本面事实并不存在，那么这个论点仍然是不完整的。

这个功能的另一个优点是它会同时返回一个简短的标签和相应的解释原因。这样一来，最终的输出结果就更容易被理解了；同时，对于编写备忘录的人来说，这样的格式也比单纯的分类方式更加便于使用。

构建事实对象

在开始撰写备忘录之前，系统会首先将所有相关信息整合成一个结构化的对象。这个对象将成为最终输出结果的唯一依据。我们不会向模型提供分散的各类数据，而是会提供一个包含论点、各种信号指标、公司背景信息、证据以及最终判断结果在内的完整资料包。

1. 公司背景信息

首先，我们会使用一个小工具从基础数据中提取公司的基本背景信息。

将其添加到core.py文件中：

def extract_company_context(fundamentals):
    if not isinstance(fundamentals, dict):
        return {}

    gen = fundamentals.get("General", "") or {}

    out = {
        "name": gen.get("Name"),
        "code": gen.get("Code"),
        "exchange": gen.get("Exchange"),
        "sector": gen.get("Sector"),
        "industry": gen.get("Industry"),
        "country": gen.get("CountryName"),
        "market_cap": gen.get("MarketCapitalization"),
        "pe_ratio": gen.get("PERatio"),
        "beta": gen.get("Beta"),
        "dividend_yield": gen.get("DividendYield"),
        "description": gen.get("Description"),
    }

    clean = {}
    for k, v in out.items():
        if v not in (None, "", "NA"):
            clean[k] = v

    return clean

这个函数只是一个整理数据的步骤。它能够为我们提供一个简洁的公司背景信息模块，这样在后续的分析中，我们就可以将这些信息与价格数据及各种基本面指标一起使用，而无需将整个基础数据包都纳入备忘录的编写流程中。

2. 单股事实信息构建器

现在再添加单股事实信息构建器：

def build_thesis_facts(parsed, ticker, signals, fundamentals, thesis_tags, evidence):
    company = extract_company_context(fundamentals)

    facts = {
        "type": "single_name_thesis_test",
        "ticker": ticker,
        "lookback_days": parsed["lookback_days"],
        "thesis": parsed["thesis"],
        "thesis_summary": thesis_tags.get("summary", ""),
        "claim_types": thesis_tags.get("claim_types", []),
        "marketsignals": {
            "ret_total": signals.get("ret_total"),
            "vol_annualized": signals.get("vol_annualized"),
            "max_drawdown": signals.get("max_drawdown"),
            "trend_slope": signals.get("trendslope"),
            "ret_to_vol": signals.get("ret_to_vol"),
            "start_price": signals.get("start_price"),
            "end_price": signals.get("end_price"),
            "n_points": signals.get("n_points"),
        },
        "company_context": {
            "name": company.get("name"),
            "exchange": company.get("exchange"),
            "sector": company.get("sector"),
            "industry": company.get("industry"),
            "country": company.get("country"),
            "market_cap": company.get("market_cap"),
            "pe_ratio": company.get("pe_ratio"),
            "beta": company.get("beta"),
            "dividend_yield": company.get("dividend_yield"),
        },
        "description": company.get("description"),
        "evidence_for": evidence.get("evidence_for", []),
        "evidence_against": evidence.get("evidence_against",`),
        "missing_evidence": evidence.get("missing_evidence", ""),
    }

    facts["verdict"] = decide_verdict(evidence, thesis_tags.get("claim_types", []))
    return facts

这是针对单只股票的研究报告所使用的主要数据结构。它整合了解析后的研究报告内容、市场信号数据、公司基本情况信息、各类证据资料以及最终结论。在这个阶段，辅助系统已经完成了所有的推理工作；这份备忘录并不会产生任何新的结论，而是仅仅基于这些数据来进行编写。

3. 关注列表数据构建器

现在我们来添加针对关注列表的数据处理功能：

def build_watchlist_facts(parsed, tickers, signals_by_ticker, fundamentals_by_ticker, thesis_tags, evidence_by_ticker):
    per_ticker = {}

    for t in tickers:
        company = extract_company_context(fundamentals_by_ticker.get(t, {}))
        signals = signals_by_ticker.get(t,{})
        evidence = evidence_by_ticker.get(t, {})

        per_ticker[t] = {
            "company_context": {
                "name": company.get("name"),
                "sector": company.get("sector"),
                "industry": company.get("industry"),
                "market_cap": company.get("market_cap"),
                "pe_ratio": company.get("pe_ratio"),
                "beta": company.get("beta"),
            },
            "marketsignals": {
                "ret_total": signals.get("ret_total"),
                "vol_annualized": signals.get("vol_annualized"),
                "max_drawdown": signals.get("max_drawdown"),
                "trend_slope": signals.get("trendslope"),
                "ret_to_vol": signals.get("ret_to_vol"),
            },
            "evidence_for": evidence.get("evidence_for", []),
            "evidence_against": evidence.get("evidence_against", []),
            "missing_evidence": evidence.get("missing_evidence", ""),
            "verdict": decide_verdict(evidence, thesis_tags.get("claim_types", []))
        }

    facts = {
        "type": "watchlist_thesis_test",
        "tickers": tickers,
        "lookback_days": parsed["lookback_days"],
        "thesis": parsed["thesis"],
        "thesis_summary": thesis_tags.get("summary", ""),
        "claim_types": thesis_tags.get("claim_types", []),
        "per_ticker": per_ticker,
    }

    return facts

这个版本的功能与之前的相同，只不过它是针对多个股票来处理的。它没有使用一个顶级的证据数据结构，而是为每只股票分别存储相关数据，这样在编写备忘录时就可以直接比较各只股票的信息，而无需重新构建数据结构。

这就是这一部分内容之所以重要的原因。当我们进入备忘录编写阶段时，我们不再希望传递一些松散、无结构的数据；我们需要的是一个已经包含了以下所有信息的结构化对象：

  • 研究报告内容

  • 相关的市场信号数据

  • 公司的基本情况信息

  • 各类证据资料

  • 最终结论

这样的结构使得最后的编写步骤更加清晰明了，也使整个工作流程更易于调试。

合理性检查（Jupyter Notebook)

在test.ipynb文件中运行这段代码，就可以快速进行合理性检查了：

from core import build_thesis_facts, extract_company_context

facts = build_thesis_facts(
    parsed={
        "tickers": ["AAPL"],
        "lookback_days": 180,
        "thesis": "苹果公司具有吸引力，因为其下行风险已被控制，且业务质量依然很高。",
        "mode": "single"
    },
    ticker="AAPL.US",
    signals=signals,
    fundamentals=funds,
    thesis_tags=tags,
    evidence=evidence
)

print(json.dumps(facts, indent=2))

预期输出：

{
  "type": "single_name_thesis_test",
  "ticker": "AAPL.US",
  "lookback_days": 180,
  "thesis": "苹果公司具有吸引力，因为其下行风险已被控制，且业务质量依然很高。",
  "thesis_summary": "苹果公司的吸引力在于其下行风险得到了有效控制，同时业务质量也非常优秀。",
  "claim_types": [
    "controlled_downside",
    "business_quality"
  ],
  "marketsignals": {
    "ret_total": -0.05675067340688533,
    "vol_annualized": 0.2504818805125429,
    "max_drawdown": -0.11322450740687473,
    "trend_slope": -0.0005437843809243782,
    "ret_to_vol": -0.22656598270006817,
    "start_price": 271.01,
    "end_price": 255.63,
    "n_points": 62
  },
  "company_context": {
    "name": "苹果公司",
    "exchange": "纳斯达克",
    "sector": "科技行业",
    "industry": "消费电子领域",
    "country": "美国",
    "market_cap": null,
    "pe_ratio": null,
    "beta": null,
    "dividend_yield": null
  },
  "description": "苹果公司负责设计、制造并销售智能手机、个人电脑、平板电脑、可穿戴设备以及各类配件。该公司推出的产品包括iPhone智能手机系列、Mac个人电脑系列、iPad多功能平板电脑系列，还有AirPods、Apple Vision Pro、Apple TV、Apple Watch等可穿戴设备及家居用品，同时还提供各种苹果品牌及第三方品牌的配件。此外，苹果公司还提供AppleCare技术支持和云服务，并运营着App Store等平台，让消费者能够下载应用程序和数字内容，如书籍、音乐、视频、游戏和播客等；同时也通过广告服务进行盈利，包括与第三方合作进行的授权业务以及自己拥有的广告平台。另外，苹果公司还提供多种订阅服务，比如Apple Arcade游戏订阅服务、Apple Fitness+个性化健身服务、Apple Music定制音乐播放服务、Apple News+新闻杂志订阅服务、Apple TV独家原创内容及直播体育赛事服务等。苹果公司还发行了联名信用卡Apple Card，并提供了无现金支付服务Apple Pay，同时还授权他人使用自己的知识产权。苹果公司的客户群体包括个人消费者、中小型企业，以及教育机构、企业和政府部门。该公司通过App Store向用户提供第三方应用程序；同时也通过自己的零售店、在线商店和直销团队销售产品，此外还通过与第三方移动网络运营商的合作进行产品销售。苹果公司前身为Apple Computer, Inc.，并于2007年1月更名为苹果公司。苹果公司成立于1976年，总部位于加利福尼亚州的库比蒂诺市。",
  "evidence_for": [
    "最大下行风险仅为-11.32%，处于相对可控的范围之内。"
  ],
  "evidence_against": [],
  "missing_evidence": [
    "该分析版本没有包含利润率、增长速度、现金流或资本回报率等直接反映业务质量的数据。",
    "仅提供了基本的公司背景信息，这些信息不足以单独用来评估其业务质量。",
    "没有相关的β值数据。"
  ],
  "verdict": {
    "verdict": "部分得到支持",
    "reason": "部分论点得到了验证，但缺乏直接反映业务质量的证据。"
  }
}

撰写最终备忘录

到这个阶段为止，最困难的部分已经完成了。

当我们进入撰写备忘录的环节时，辅助系统就已经生成了一个结构化的数据对象，其中包含了论点、证据类型、市场信号、公司背景信息、相关证据以及最终结论。因此，这份最终备忘录并不是进行推理分析的地方，它仅仅是一个将结构化判断结果转化为可读文本的呈现工具而已。

请将以下代码添加到core.py文件中：

def write_thesis_memo(facts):
    prompt = f"""
您正在撰写一份简短的金融研究备忘录。
请仅使用下面提供的事实来进行编写。
不要凭空编造任何数字、事件、对比数据或观点；
如果缺少某些证据，也请明确说明。
请严格按照以下结构进行编写：
1. 需要审查的论点
2. 支持该论点的证据
3. 反驳该论点的证据
4> 缺失的证据
5. 最终结论
6. 总体评估

格式要求：
- 保持简洁明了
- 采用分析性且专业的表达方式
- 除非必要，否则不要使用项目符号
- 避免夸大其词或使用泛泛的投资风险警示语句
- 总体评估应当基于客观证据，并保持中立公正
- 最终结论部分必须明确使用所提供的结论

输入数据：
{json.dumps(facts, indent=2, default=str)}
"".strip()

    r = oaresponses.create(
        model=model_name,
        input=[{"role": "user", "content": prompt}],
    )

    return r.output_text.strip()

这个函数将模型的功能限制在一个具体的任务范围内：它不需要分析原始的价格历史数据、基本财务信息或各种分散的变量，而是直接根据已经包含判断结果的结构化数据对象来撰写备忘录。

这种分离机制非常重要，因为它能确保最终生成的备忘录内容始终基于客观事实。模型并不会在最后一刻才决定自己对这只股票的看法，它只是将前面步骤产生的结构化输出结果整理成一份简短的研究报告而已。

此外，提示语也经过了精心设计，它明确了备忘录的编写结构，要求模型不得随意添加任何内容，并且要求最终结论必须明确表述，而不能含糊其辞。这些规定有助于确保即使基础论点发生变化，最终的输出结果也能保持一致性。

合理性检查（Jupyter Notebook)

您可以使用前面章节中的数据对象来测试这个功能：

from core import write_thesis_memo

memo = write_thesis_memo(facts)
print(memo)

预期输出结果：
合理性检查的预期输出结果

将所有部分整合起来

现在，所有的组件都已经准备好了。我们有了解析器、数据采集工具、信号生成系统、论点分类器、证据处理模块、结论生成层以及备忘录编写功能。接下来要做的就是将这些组件连接成一个完整的端到端流程。

将以下代码添加到 `core.py` 文件中：

async def run_thesis_copilot(user_text):
    trace_id = uuid.uuid4().hex[:10]
    log_event("request_started", trace_id, text=user_text)

    parsed = enforce_limits(parse_request(user_text))
    tickers = parsed["tickers"]

    if not tickers:
        return {
            "memo": "请求中未找到有效的股票代码。",
            "facts": {},
            "data_used": {},
            "tool_trace_id": trace_id,
        }

    log_event(
        "parsed",
        trace_id,
        tickers=tickers,
        lookback_days=parsed["lookback_days"],
        mode=parsed["mode"],
        thesis=parsed["thesis"],
    )

    start_date, end_date = get_dates_from_lookback(parsed["lookback_days"])
    state = make_state()

    try:
        thesis_tags = classify_thesis(parsed["thesis"])

        if parsed["mode"] == "single":
            ticker = tickers[0]
            ticker_full = ticker if "." in ticker else f"{ticker}.US"

            log_event(
                "tool_phase",
                trace_id,
                mode="single",
                ticker=ticker_full,
                start_date=start_date,
                end_date=end_date,
            )

            prices = await fetch_prices(ticker_full, start_date, end_date, trace_id, state)
            funds = await fetch_fundamentals(ticker_full, trace_id, state)

            pricesignals = compute_price_signals(prices)
            fundamental Signals = compute_fundamentalSignals(funds)

            evidence = build_evidence_blocks(
                parsed["thesis"],
                thesis_tags,
                pricesignals,
                fundamental Signals
            )

            facts = build_thesis_facts(
                parsed,
                ticker_full,
                pricesignals,
                funds,
                thesis_tags,
                evidence
            )

            facts["fundamental_signals"] = fundamentalSignals

            memo = write_thesis_memo(facts)

            out = {
                "memo": memo,
                "facts": facts,
                "data_used": {
                    "tickers": [ticker_full],
                    "date_range": [start_date, end_date],
                    "tools_called": [x.get("tool") for x in state["tool_trace"],
                    "tool_calls": state["tool_calls"],
                },
                "tool_trace_id": trace_id,
            }

            log_event("request_finished", trace_id, tool_calls=state["tool_calls"])
            return out

        ticker_full = [x if "." in x else f"{x}.US" for x in tickers]

        log_event(
            "tool_phase",
            trace_id,
            mode="watchlist",
            tickers=ticker_full,
            start_date=start_date,
            end_date=end_date,
        )

        signals_by_ticker = {}
        funds_by_ticker = {}
        evidence_by_ticker = {}

        for t in ticker_full:
            prices = await fetch_prices(t, start_date, end_date, trace_id, state)
            funds = await fetch_fundamentals(t, trace_id, state)

            pricesignals = compute_price_signals(prices)
            fundamental Signals = compute_fundamentalSignals(funds)

            evidence = build_evidence_blocks(
                parsed["thesis"],
                thesis_tags,
                pricesignals,
                fundamental Signals
            )

            signals_by_ticker[t] = {
                **price_signals,
                "fundamental_signals": fundamental Signals
            }
            funds_by_ticker[t] = funds
            evidence_by_ticker[t] = evidence

        facts = build_watchlist_facts(
            parsed,
            ticker_full,
            signals_by_ticker,
            funds_by_ticker,
            thesis_tags,
            evidence_by_ticker,
        )

        memo = write_thesis_memo(facts)

        out = {
            "memo": memo,
            "facts": facts,
            "data_used": {
                "tickers": ticker_full,
                "date_range": [start_date, end_date],
                "tools_called": [x.get("tool") for x in state["tool_trace"],
                "tool_calls": state["tool_calls"],
            },
            "tool_trace_id": trace_id,
        }

        log_event("request_finished", trace_id, tool_calls=state["tool_calls"])
        return out

    except Exception as e:
        detail = repr(e)
        if hasattr(e, "exceptions"):
            detail = detail + " | " + " ; ".join([repr(x) for x in e.exceptions])

        log_event("request_failed", trace_id, err=detail)

        return {
            "memo": f"失败原因：{e}",
            "facts": {},
            "data_used": {
                "tickers": tickers,
                "date_range": [start_date, end_date],
                "tools_called": [x.get("tool") for x in state["tool_trace"],
                "tool_calls": state["tool_calls"],
            },
            "tool_trace_id": trace_id,
        }

这个功能实际上将整个分析流程整合在了一个地方。它负责解析用户输入的请求，获取所需数据，计算出两个相关的信号指标，构建相应的分析依据，整理出所有相关事实，生成分析报告，并最终以结构清晰的形式输出所有结果。

它的优点在于输出的不仅仅是一份分析报告。同时还会提供结构化的事实信息、所使用的分析工具、时间范围以及追踪编号等信息。这些信息使得最终的分析结果具有可核查性，从而避免了让这个辅助系统变成一个“黑箱”。

演示时间！（Jupyter笔记本格式）

演示1：验证高端收费是否合理

这个演示案例非常合适，因为它让这个辅助系统超越了单纯对比两家公司基本情况的范畴。系统询问的并不是NVIDIA是否是一家优秀的公司，而是NVIDIA相对于AMD的高端收费是否能够通过市场表现和业务质量来得到合理解释。

以下是具体的指令代码：

from core import run_thesis_copilot

q = """
在NVDA和AMD之间，我认为由于NVDA的市场表现和业务质量更优秀，因此其高端收费是合理的。
请验证过去6个月内的情况。
"".strip()

result = await run_thesis_copilot(q)

print(result["memo"])
print(result["data_used"])

而输出结果如下：

演示1的输出结果

这种输出方式的好处在于，它并不会简单地将分析结果归纳为“是”或“否”。显然，NVIDIA在业务质量方面表现更佳，但市场表现则不够令人信服；同时，由于缺乏直接的估值数据，这个辅助系统也不会做出夸大其词的结论。

这正是我们所需要的分析方式——该系统并不是单纯地对比两家公司，而是会验证关于高端收费合理性的具体论点是否成立。

演示2：判断股票波动性是否过高

第二个演示案例仍然围绕单只股票展开分析，但探讨的角度不同。这次系统询问的不是这家公司是否有投资价值，而是该股票的波动性是否超出了其业务质量所能够支撑的范围。

以下是具体的指令代码：

q = """
TSLA的波动性似乎过高，与其业务质量不相符。
请验证过去一年内的情况。
"".strip()

result = await run_thesis_copilot(q)

print(result["memo"])
print(result["data_used"])

而输出结果如下：

演示2的输出结果这一结果颇具参考价值，因为它揭示了一个充满矛盾的观点：特斯拉近期的业绩表现以及人们对其未来发展的预期确实为这一观点提供了一定的支持，但目前的盈利能力、近期运营状况的变化、各项数据的修正值，以及股价的波动性，都表明该公司的业务质量并不足以完全支撑其所承担的那些风险。

因此，最终的结论落在了它应该所在的位置：并非一个确凿无疑的结论，而只是一个缺乏充分支持的论点。

最后的思考

目前而言，这个辅助系统已经能够很好地完成其中最重要的任务。它可以将自然语言形式的论点通过EODHD的MCP层提取出相关的市场数据及基本财务信息，将这些原始资料转化为结构化的数据，从而生成一份比普通的股票分析报告更加严谨的研究报告。

不过，这一版本仍然存在明显的局限性：它尚未深入探讨会计报表中的逻辑关系，也没有结合新闻事件或各种影响因素来进行分析；在处理相对估值问题时，对于那些要求更高的比较案例来说，其分析能力仍有进一步提升的空间。

但即便存在这些局限，这种变革已经具有重要的意义。真正的变化并不在于将某个模型与财务数据连接起来，而在于从单纯的股票信息汇总转向对各种论点的实质性验证。

Related Posts:

Tags:
Comments are closed.