最近 AI 实在太火了。

这两年市场上最强的主线，绕来绕去基本都绕不开 AI。英伟达、AMD、博通、台积电、美光、微软、谷歌、Meta、CoreWeave、Supermicro……每隔一段时间，总有一家公司因为财报、订单、资本开支、HBM、GPU、云计算、数据中心这些关键词突然大涨。

更有意思的是，它们往往不是孤立上涨。

有时候是英伟达先动，然后市场开始挖 ASIC、光模块、服务器、液冷、电力；有时候是美光财报超预期，随后资金又去看 HBM、存储、半导体设备、先进封装；还有时候是云厂商上调资本开支，整个 AI 基础设施链都会被重新定价。

这更像一种 AI 主题内部的资产轮动。市场不是简单地买一只股票，而是在一个大叙事下面不断寻找新的分支、新的解释和新的补涨对象。

问题也就来了：这些关系如果全靠人工整理，太累了。

一个人可以知道英伟达、AMD、美光、微软、谷歌，但很难持续跟踪几十上百个 TradFi 股票合约的业务变化、新闻变化、板块归属和股价联动。更麻烦的是，很多公司不是单一属性。谷歌既是 AI 云，又是 Gemini，又是广告平台；微软既是云，又是 Copilot，又是企业软件；博通既有 ASIC，又有网络芯片，还和 AI 数据中心有关。

所以我开始尝试：能不能让大模型帮我们搭一张 AI 股票图谱？

不是让大模型直接告诉我们买什么，而是让它做它更擅长的事：理解公司业务，归类 AI 子板块，整理新闻，解释事件，判断一家公司异动可能影响哪些同主题股票。真正的交易决策，还是交给价格、统计和风控。

先把股票放进正确的 basket

策略的第一步不是交易，而是建图谱。

交易所现在有不少 TradFi 股票合约，可以直接拿到一批类似 NVDA、AMD、MU、MSFT、GOOGL、SMCI 这样的股票合约。系统先筛选股票类合约，只保留真正的 EQUITY 标的。

这里有一个很重要的细节：不能把 ETF、指数、杠杆 ETF 混进来。比如 QQQ、SOXL、EWT 这类东西，虽然也可能出现在合约列表里，但它们不是公司股票。如果把它们放进 AI 股票 basket，后面的统计关系会被污染。

核心筛选逻辑大概是这样：

def refresh_equity_universe(store, force_reanalyze):
    ms = exchange.GetMarkets()
    symbols = []

    for key, market in ms.items():
        info = market.get("Info", {}) or {}
        sub_type = info.get("underlyingSubType", []) or []
        underlying_type = info.get("underlyingType", "")

        if ".swap" not in key:
            continue
        if "TradFi" not in sub_type:
            continue
        if underlying_type != "EQUITY":
            continue

        symbol = normalize_symbol(key)
        store["markets"][symbol] = {
            "symbol": symbol,
            "contract": key,
            "underlyingType": underlying_type,
            "amountPrecision": market.get("AmountPrecision", 0),
            "pricePrecision": market.get("PricePrecision", 2),
            "ctVal": market.get("CtVal", 1) or 1,
            "minQty": market.get("MinQty", 0) or 0,
        }
        symbols.append(symbol)

拿到股票池后，系统会让大模型给每只股票做画像。

画像不是简单写一句“这是科技股”，而是要判断它属于哪些 AI 子板块。比如一只股票可以同时属于多个 basket。谷歌可以属于 AI 云、AI 模型应用、AI 平台广告；微软可以属于 AI 云和企业 AI 应用；英伟达可以属于 AI 算力和 AI 服务器；美光可以属于 AI 存储。

这一步的重点是：AI 时代的股票不是单标签的，而是多标签的。

每只股票都带有几个属性：属于哪些 basket、暴露度有多高、在这个 basket 里更像 leader 还是 follower、可能受哪些公司影响、又可能影响哪些公司。

我给大模型的输出结构做了强约束：

schema = {
    "symbol": symbol,
    "asset_type": "stock|adr|etf|leveraged_etf|index|fund|other",
    "company_name": "",
    "sector": "",
    "business_summary": "",
    "tradable": False,
    "model_confidence": 0,
    "baskets": [
        {
            "id": "ai_cloud",
            "exposure": 0.0,
            "role": "leader|follower|both|observer",
            "sub_theme": "",
            "reason": ""
        }
    ],
    "search_names": [],
    "event_keywords": [],
    "negative_keywords": [],
    "possible_leaders": [],
    "possible_followers": [],
    "brave_queries": [],
    "reason": ""
}

这里最关键的是 asset_type。只有 stock 和 adr 可以进入 basket，ETF、杠杆 ETF、指数、基金、商品或地区 ETF 都直接过滤掉。

另外，reason 必须用中文，字段名必须一致，置信度统一成 0 到 100。否则后面实盘页面看起来会很乱。

板块信息不能靠脑补

光有股票分类还不够。

AI 板块变化太快了。今天市场炒 GPU，明天可能炒 HBM，后天可能炒光模块，再过几天又开始炒液冷、电力、云厂商资本开支。

如果只靠大模型自己的知识，很容易过时。所以系统还需要实时信息源。

我这里接入了 BraveSearch。系统会定期搜索每个 AI basket 的板块信息，也会搜索 basket 内重点股票的新闻。

比如 AI 存储 basket，会去看美光、西部数据、SanDisk、HBM、DRAM、NAND 这些关键词；AI 光通信 basket，会看 Ciena、Credo、Lumentum、AAOI、800G、1.6T 光模块这些信息。

板块情报的核心流程是：

def refresh_basket_intelligence(store):
    if not BRAVE_KEY:
        return
    if not LLM_API_KEY:
        return

    intel = {}

    for bid, basket in store["baskets"].items():
        members = sorted(
            basket["members"],
            key=lambda x: x.get("exposure", 0),
            reverse=True
        )
        top_symbols = [m["symbol"] for m in members[:8]]

        basket_news = fetch_basket_news(basket, top_symbols)
        stock_news = {}

        for symbol in top_symbols[:6]:
            profile = store["profiles"].get(symbol, {})
            stock_news[symbol] = fetch_stock_news(profile, bid)
            Sleep(800)

        analysis = analyze_basket_intel_with_llm(
            store, basket, top_symbols, basket_news, stock_news
        )

        if analysis:
            analysis["basket"] = bid
            analysis["basket_name"] = basket["name"]
            analysis["symbols"] = top_symbols
            analysis["updatedAt"] = int(time.time())
            intel[bid] = analysis

    store["basket_intel"] = intel

这一步的作用是给每个 AI 子板块建立一个动态背景。

大模型会根据 BraveSearch 拉到的新闻，输出板块趋势、核心催化、核心风险、候选 leader、候选 follower，以及后续需要重点验证的关系。

这和直接问大模型“AI 存储怎么看”不一样。我们先把实时新闻喂给它，再让它总结。

不是新闻驱动交易，而是价格先异常

一开始我也想过新闻驱动。

比如看到某个公司财报超预期，就让大模型判断利好哪些股票，然后直接生成信号。但实际想想，这条路噪音很大。新闻太多了，很多新闻看起来很重要，但市场根本不买账。

所以我把逻辑反过来了。

先看价格，再看新闻。

如果一个 basket 里所有股票都比较平稳，系统就只是观察。只有当某只股票明显偏离板块平均表现，成为当前 leader，系统才会去追问：它为什么动？

这一步用的是 basket 内部的 ZScore。

简单理解，就是看某只股票相对同板块其他股票是不是异常强或异常弱。如果一个股票突然跑出来，而同 basket 里的其他股票还没同步反应，就可能存在扩散机会。

计算 basket ZScore 的核心逻辑如下：

def calc_basket_zscores(store, symbols):
    returns = {}
    min_len = 999999

    for symbol in symbols:
        bars = get_records(
            store["markets"][symbol]["contract"],
            KLINE_PERIOD,
            SYNC_WINDOW + 10
        )
        if len(bars) < SYNC_WINDOW + 2:
            return None

        r = calc_returns(bars)[-SYNC_WINDOW:]
        returns[symbol] = r
        min_len = min(min_len, len(r))

    latest = {s: returns[s][-1] for s in symbols}
    avg_latest = mean(list(latest.values()))
    dev_series = {s: [] for s in symbols}

    for i in range(min_len):
        row = [returns[s][-min_len + i] for s in symbols]
        row_avg = mean(row)
        for s in symbols:
            dev_series[s].append(returns[s][-min_len + i] - row_avg)

    zscores = {}
    max_abs = 0
    for s in symbols:
        sigma = std(dev_series[s])
        z = (latest[s] - avg_latest) / sigma if sigma > 0 else 0
        zscores[s] = round(z, 3)
        max_abs = max(max_abs, abs(z))

    return {"zscores": zscores, "maxAbs": max_abs}

如果某个 basket 的最大偏离超过突破阈值，就进入 BREAKOUT_DETECTED 状态。

这时候系统不会马上交易，而是开始追因。

新闻只负责解释，统计关系负责验证

当 leader 出现价格突破后，系统会抓取它的最新新闻，再让大模型判断这条新闻是否能解释价格异动。

但这里不是简单问“这条新闻利好吗”。

系统会把当前 basket、板块情报、实时价格状态、leader 的 ZScore、候选 follower、已经验证过的统计关系、BraveSearch 拉到的新闻，一起交给大模型。

大模型需要回答：

（1）这条新闻是否能解释 leader 的价格异常。
（2）新闻方向是否和价格方向一致。
（3）是否允许交易。
（4）影响哪些 follower。
（5）每个 follower 的 impact 有多大。
（6）为什么这些 follower 可能受影响。

核心追因逻辑如下：

def explain_breakout_with_news(store, bid, leader, direction, leader_z, followers, edges):
    if not BRAVE_KEY:
        return None
    if not LLM_API_KEY:
        return None

    profile = store["profiles"].get(leader, {})
    news = fetch_stock_news(profile, bid)
    if not news:
        return None

    event = analyze_breakout_news_with_llm(
        store, bid, leader, direction, leader_z,
        followers, edges, news
    )
    if not event:
        return None

    if not event.get("explains_price_move"):
        return None

    expected_direction = "bullish" if direction == "long" else "bearish"
    if event.get("direction") != expected_direction:
        return None

    event["confidence"] = normalize_confidence(event.get("confidence", 0))
    if event["confidence"] < MIN_NEWS_CONFIDENCE:
        return None

    if not event.get("trade_allowed"):
        return None

    return event

但即使大模型说某个 follower 可能受影响，也不能马上交易。

还要看 K 线统计关系。

系统会在每个 basket 内计算股票之间的历史联动，比如 source 突破后 target 是否经常跟随，相关性如何，最佳滞后周期是多少，平均跟随收益怎么样。

只有统计关系通过的 source -> target，才会变成 active edge。

def rebuild_basket_stats(store):
    edges = []

    for bid, basket in store["baskets"].items():
        symbols = [
            m["symbol"] for m in basket["members"]
            if m["symbol"] in store["markets"]
        ]

        for source in symbols:
            for target in symbols:
                if source == target:
                    continue

                metric = validate_pair(store, source, target)
                if not metric:
                    continue

                source_exp = basket_member_info(basket, source).get("exposure", 0)
                target_exp = basket_member_info(basket, target).get("exposure", 0)

                score = clamp(
                    metric["corr"] * 0.3 +
                    metric["follow"] * 0.4 +
                    min(source_exp, target_exp) * 0.3,
                    0, 1
                )

                edge = {
                    "basket": bid,
                    "source": source,
                    "target": target,
                    "corr": round(metric["corr"], 3),
                    "follow": round(metric["follow"], 3),
                    "lag": metric["lag"],
                    "avgReturn": round(metric["avgReturn"], 6),
                    "sample": metric["sample"],
                    "score": round(score, 3),
                    "active": (
                        score >= MIN_EDGE_SCORE and
                        metric["corr"] >= MIN_CORR and
                        metric["follow"] >= MIN_FOLLOW_SUCCESS
                    ),
                    "updatedAt": int(time.time()),
                }
                edges.append(edge)

    store["edges"] = edges

这一步很重要。

大模型擅长理解逻辑，但市场不一定按逻辑走。统计关系就是用来验证：这个逻辑过去有没有被市场交易过。

交易的是扩散，不是追高

这个策略不是为了追 leader。

leader 已经突破了，直接追进去容易变成追高。策略真正想做的是同一 AI basket 内的扩散机会。

也就是：leader 已经动了，新闻也解释得通，但某些 follower 还没有完全反应。如果这些 follower 和 leader 在历史上确实有联动关系，就可能存在短期跟随机会。

突破扫描里会先找 active edge，再找滞后的 follower：

def scan_price_breakouts(store):
    for bid, rt in store["realtime"].items():
        if rt["state"] != "BREAKOUT_DETECTED":
            continue

        leader = rt["leader"]
        z = 0
        for m in rt["members"]:
            if m["symbol"] == leader:
                z = m["z"]

        direction = "long" if z > 0 else "short"

        edges = [
            e for e in store["edges"]
            if e["active"] and e["basket"] == bid and e["source"] == leader
        ]

        followers = []
        for e in edges:
            target_rt = find_realtime_member(rt, e["target"])
            if target_rt and abs(target_rt["z"]) < BREAKOUT_THRESHOLD * 0.75:
                followers.append(e["target"])

        if not followers:
            continue

只有当新闻追因也通过后，才会生成 confirmed signal。

交易对象主要是 follower。

如果当前没有持仓，就按信号方向开仓。如果同一个合约已经有同向持仓，就加仓。如果已经有反向持仓，就先平仓，再按新方向开仓。

def execute_symbol_trade(store, signal, symbol, detail):
    direction = signal["direction"]

    if TRADE_MODE != "trade":
        return open_or_add_position(store, signal, symbol, detail, add_mode=False)

    pos = get_position_by_symbol(symbol)
    if pos:
        pos_dir = position_direction(pos)

        if pos_dir == direction:
            return open_or_add_position(store, signal, symbol, detail, add_mode=True)

        close_existing_position(store, symbol, pos, "reverse_signal")
        Sleep(1000)

    return open_or_add_position(store, signal, symbol, detail, add_mode=False)

这也是因为很多股票会同时出现在多个 basket 里。一个股票可能在 AI 云里是 follower，在 AI 应用里又是另一个逻辑的成员。不同 basket 的信号可能同时影响它，所以需要统一处理同一合约的方向。

风控先做简单，但必须有

风控方面，目前用的是硬止损和移动止损。

亏损达到阈值就退出；盈利达到一定幅度后启动移动止损，如果从最高浮盈回撤过多，就锁定利润离场。

def monitor_positions(store):
    for pos in exchange.GetPosition() or []:
        symbol = position_symbol(pos)
        if symbol not in store["markets"]:
            continue

        state = _G(position_state_key(symbol)) or {}
        market = store["markets"][symbol]
        ticker = exchange.GetTicker(market["contract"])

        direction = position_direction(pos)
        pnl_pct = (
            (ticker["Last"] - pos["Price"]) /
            pos["Price"] * 100 *
            (1 if direction == "long" else -1)
        )

        if pnl_pct > state.get("peakPnlPct", 0):
            state["peakPnlPct"] = pnl_pct

        if not state.get("trailActive") and state.get("peakPnlPct", 0) >= TRAIL_ACTIVATE_PCT:
            state["trailActive"] = True

        hard_stop = False
        if state.get("stop"):
            hard_stop = (
                ticker["Last"] <= state["stop"]
                if direction == "long"
                else ticker["Last"] >= state["stop"]
            )

        trail_hit = False
        if state.get("trailActive"):
            giveback = state.get("peakPnlPct", 0) - pnl_pct
            allowed = max(
                1.5,
                state.get("peakPnlPct", 0) * TRAIL_GIVEBACK_PCT / 100.0
            )
            trail_hit = giveback >= allowed

        if hard_stop or trail_hit:
            close_existing_position(
                store, symbol, pos,
                "hard_stop" if hard_stop else "trailing_stop"
            )

这不是最终版风控，但作为第一版实盘观察已经够用。

后面还需要加账户级总仓位、单板块敞口、单日亏损限制、成交确认和加仓次数限制。

现在这个系统做到什么程度了

现在这个系统已经能完成一条比较完整的研究闭环。

它可以自动扫描 TradFi 股票合约，自动让大模型给股票做多 basket 画像，自动构建 AI 子板块，自动通过 BraveSearch 拉取板块和个股信息，自动总结板块情报，自动检测每个 basket 的实时强弱，自动统计 basket 内股票的跟随关系，并在 leader 价格突破后进行新闻追因。

如果新闻解释成立、方向一致、统计关系也支持，系统会生成 confirmed signal。默认模式下只提示，不实盘；切到 trade 模式后才会交易。

实盘页面上也不再只是日志，而是把系统概览、basket 实时状态、板块情报、最近信号、最近交易、最近步骤用表格展示出来。这样可以比较清楚地看到系统卡在哪一步：是只有价格异常，还是新闻没有解释，还是 follower 不满足条件，还是风控拦截。

现在还不成熟的地方

实话说，这个系统还不能算成熟自动交易策略。

它更像是一个 AI 主题轮动研究框架，已经能把“股票画像、板块情报、价格异常、新闻追因、统计验证、交易信号”串起来，但还需要继续验证。

最需要继续完善的是几块。

（1）股票画像质量。大模型有时会把一些边缘股票放进 basket，所以必须强制 asset_type，避免 ETF、指数、地区基金混入。

（2）新闻质量。BraveSearch 抓到的新闻不一定都是核心新闻，模型需要判断哪些能真正解释价格。

（3）统计关系。历史跟随关系不等于未来一定有效，需要持续复盘。

（4）交易风控。自动 trade 前，必须补账户级仓位、板块敞口、日亏损和成交确认。

所以现在更适合先跑 notify 模式，看 confirmed signal 的质量。

如果一段时间后发现信号确实能解释市场异动，并且 follower 后续经常有反应，再考虑小仓位实盘。

写在最后

这个策略不是让 AI 直接下单。

我一直觉得，大模型在交易系统里最适合的位置，不是“告诉我买什么”，而是帮我们处理人工很难持续维护的信息结构：公司业务、板块归属、实时新闻、事件解释、资金轮动路径、潜在影响对象。

真正的交易还是要交给价格、统计和风控。

AI 时代的量化策略，不应该只是多接一个大模型接口，而是要把实时数据、主题图谱、新闻事件、价格结构、交易执行和复盘解释串起来。

这个 AI 股票图谱策略，就是沿着这个方向做的一次尝试。