最近AIが本当に熱い。

この2年間、市場で最も強いテーマは、何と言ってもAIから逃れられない。NVIDIA、AMD、Broadcom、TSMC、Micron、Microsoft、Google、Meta、CoreWeave、Supermicro……しばらくすると、必ずどこかの企業が決算、受注、設備投資、HBM、GPU、クラウドコンピューティング、データセンターといったキーワードで急騰する。

さらに興味深いのは、それらが単独で上昇するわけではないことだ。

ある時はNVIDIAが先に動き、その後市場がASIC、光モジュール、サーバー、液冷、電力に注目し始める。またある時はMicronの決算が予想を上回り、資金がHBM、ストレージ、半導体製造装置、先進パッケージングに流れる。さらにクラウド事業者が設備投資を引き上げると、AIインフラチェーン全体が再評価される。

これはどちらかと言えば、AIというテーマ内部での資産ローテーションだ。市場は単に一銘柄を買うのではなく、大きなストーリーの中で新しいセクター、新しい解釈、そして遅れて上昇する銘柄を絶えず探している。

問題は、これらの関係をすべて人手で整理するのは非常に大変だということだ。

一個人はNVIDIA、AMD、Micron、Microsoft、Googleを知っていても、数十から数百ものTradFi株式契約の事業変化、ニュース変化、セクター帰属、株価連動を継続的に追跡するのは難しい。さらに厄介なのは、多くの企業が単一の属性を持たないことだ。GoogleはAIクラウドであり、Geminiであり、広告プラットフォームでもある。Microsoftはクラウドであり、Copilotであり、エンタープライズソフトウェアでもある。BroadcomはASICを持ち、ネットワークチップも持っていて、AIデータセンターにも関連している。

そこで私は試し始めた：AIにAI株式マップを作ってもらえないか？

AIに「何を買うべきか」を直接教えてもらうのではなく、AIがより得意なことを任せる：企業の事業を理解し、AIサブセクターを分類し、ニュースを整理し、イベントを説明し、ある企業の異変が同じテーマのどの銘柄に影響を与えるかを判断する。実際の取引判断は価格、統計、リスク管理に任せる。

まず株式を正しいバスケットに入れる

戦略の最初のステップは取引ではなく、マップを作ることだ。

取引所には現在多くのTradFi株式契約があり、NVDA、AMD、MU、MSFT、GOOGL、SMCIのような株式契約を直接取得できる。システムはまず株式タイプの契約をフィルタリングし、本当のEQUITY対象のみを残す。

ここで非常に重要な詳細がある：ETF、指数、レバレッジETFを混入させてはいけない。例えばQQQ、SOXL、EWTのようなものも契約リストに現れるかもしれないが、これらは企業の株式ではない。それらをAI株式バスケットに入れると、後の統計関係が汚染される。

コアとなるフィルタリングロジックは以下のようになる：

            python
            
            
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サブセクターに属するかを判断する必要がある。例えば、1つの株式は同時に複数のバスケットに属することができる。GoogleはAIクラウド、AIモデル応用、AIプラットフォーム広告に属することができる。MicrosoftはクラウドとエンタープライズAI応用に属することができる。NVIDIAはAIコンピューティングとAIサーバーに属することができる。MicronはAIストレージに属することができる。

このステップのポイントは：AI時代の株式は単一ラベルではなく、複数ラベルを持つということだ。

各株式はいくつかの属性を持つ：どのバスケットに属するか、エクスポージャーの高さ、そのバスケット内でリーダーかフォロワーか、どの企業から影響を受ける可能性があるか、どの企業に影響を与える可能性があるか。

私は大規模言語モデルの出力構造に強い制約を設けた：

            python
            
            
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 のみがバスケットに入れることができ、ETF、レバレッジETF、指数、ファンド、コモディティ、地域ETFは直接フィルタリングされる。

さらに、reasonは中国語で記述する必要があり、フィールド名は統一しなければならない。信頼度は一律0から100にする。そうしないと、後の実践ページが乱雑に見えるからだ。

セクター情報は想像に頼ってはいけない

株式の分類だけでは不十分だ。

AIセクターの変化は非常に速い。今日市場がGPUを追うかと思えば、明日はHBM、明後日は光モジュール、その後液冷、電力、クラウド事業者の設備投資に移る。

大規模言語モデルの知識だけに頼ると、すぐに時代遅れになる。そこでシステムはリアルタイムの情報源も必要とする。

私はここでBraveSearchを組み込んだ。システムは定期的に各AIバスケットのセクター情報を検索し、バスケット内の重点株式に関するニュースも検索する。

例えば、AIストレージバスケットでは、Micron、Western Digital、SanDisk、HBM、DRAM、NANDといったキーワードを調べる。AI光通信バスケットでは、Ciena、Credo、Lumentum、AAOI、800G、1.6T光モジュールといった情報を調べる。

セクター情報のコアフローは以下の通りである：

            python
            
            
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が取得したニュースに基づいて、セクターのトレンド、中核的な触媒、中核的なリスク、リーダー候補、フォロワー候補、および今後重点的に検証すべき関係を出力します。

これは、大規模モデルに「AIストレージをどう見るか」と直接質問するのとは異なります。まずリアルタイムのニュースを入力し、その後に要約させます。

ニュース主導の取引ではなく、まず価格が異常であること

当初は私もニュース主導の方法を考えました。

例えば、ある企業の決算が予想を上回った場合、大規模モデルにどの銘柄に有利かを判断させ、そのままシグナルを生成させる方法です。しかし実際に考えると、この方法はノイズが多すぎます。ニュースは非常に多く、重要に見えても市場が全く反応しないこともあります。

そこでロジックを逆転させました。

まず価格を見て、次にニュースを見ます。

バスケット内のすべての銘柄が比較的安定している場合、システムは単に観察します。ある銘柄がセクターの平均パフォーマンスから明らかに逸脱し、現在のリーダーとなった場合にのみ、システムは「なぜ動いたのか？」と問いかけます。

このステップでは、バスケット内のZスコアを使用します。

簡単に言うと、ある銘柄が同じセクターの他の銘柄と比べて異常に強いか弱いかを確認します。ある銘柄が突然頭角を現し、同じバスケット内の他の銘柄がまだ同期して反応していない場合、拡散の機会が存在する可能性があります。

バスケットZスコア計算の中核ロジックは以下の通りです。

            python
            
            
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}

特定のバスケットの最大偏差がブレイクアウト閾値を超えた場合、BREAKOUT_DETECTED 状態に移行します。

この時点ではシステムはすぐに取引せず、原因の追及を開始します。

ニュースは説明のみを担当し、統計的関係が検証を担当する

リーダーに価格ブレイクアウトが発生した後、システムはその最新ニュースを取得し、大規模モデルにそのニュースが価格変動を説明できるかどうかを判断させます。

ただし、単に「このニュースは強気か」と尋ねるわけではありません。

システムは、現在のバスケット、セクター情報、リアルタイム価格状態、リーダーのZスコア、候補フォロワー、既に検証済みの統計的関係、BraveSearchで取得したニュースをすべて大規模モデルに渡します。

大規模モデルは以下に答える必要があります。

(1) このニュースがリーダーの価格異常を説明できるか。
(2) ニュースの方向性が価格方向と一致しているか。
(3) 取引が許可されるか。
(4) どのフォロワーに影響を与えるか。
(5) 各フォロワーへの影響度はどの程度か。
(6) なぜそれらのフォロワーが影響を受ける可能性があるのか。

中核的な原因追及ロジックは以下の通りです。

            python
            
            
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

しかし、大規模モデルがあるフォロワーが影響を受ける可能性があると言っても、すぐに取引することはできません。

Kラインの統計関係も確認する必要があります。

システムは各バスケット内で銘柄間の過去の連動を計算します。例えば、ソースがブレイクアウトした後にターゲットが頻繁に追随するか、相関性はどうか、最適なラグ期間はいくつか、平均追随リターンはどうか、などです。

統計関係が合格したソース→ターゲットのみがアクティブエッジになります。

            python
            
            
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

このステップは非常に重要です。

大規模モデルはロジックの理解に優れていますが、市場は必ずしもロジック通りに動くとは限りません。統計関係は、このロジックが過去に市場で取引されたことがあるかどうかを検証するために使用されます。

取引するのは拡散であり、高値追いではない

この戦略はリーダーを追うためのものではありません。

リーダーはすでにブレイクアウトしており、直接追いかけると高値掴みになりやすいです。戦略が本当に狙うのは、同じAIバスケット内の拡散の機会です。

つまり、リーダーは動いたが、ニュースでも説明がつくが、一部のフォロワーはまだ完全に反応していない。これらのフォロワーとリーダーが過去に実際に連動関係がある場合、短期的な追随の機会が存在する可能性があります。

ブレイクアウトスキャンではまずアクティブエッジを探し、その後遅れているフォロワーを探します：

            python
            
            
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）が生成されます。

取引対象は主にフォロワーです。

現在ポジションがない場合、シグナルの方向で新規建てを行います。同じ銘柄に同じ方向のポジションが既にある場合は、追加建てを行います。反対方向のポジションが既にある場合は、まず決済してから新しい方向で新規建てを行います。

            python
            
            
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)

これは、多くの銘柄が同時に複数のバスケットに出現するためでもあります。ある銘柄はAIクラウドではフォロワーであり、AIアプリケーションでは別のロジックのメンバーである場合があります。異なるバスケットからのシグナルが同時にその銘柄に影響を与える可能性があるため、同一銘柄の方向を統一して処理する必要があります。

リスク管理はまずシンプルに、しかし必須

リスク管理面では、現在はハードストップロスとトレーリングストップロスを使用しています。

損失が閾値に達したら退出します。利益が一定の幅に達したらトレーリングストップロスを開始し、最大含み益から過大にリトレースした場合、利益を確定して離場します。

            python
            
            
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"
            )

これは最終版のリスク管理ではありませんが、第一版の実運用観察としては十分です。

今後はアカウントレベルの総ポジション、セクターごとのエクスポージャー、1日の損失制限、約定確認、ポジション追加回数制限などを追加する必要があります。

現在このシステムはどの程度までできているか

現在、このシステムは比較的完全な研究サイクルを実行できます。

TradFi株式契約を自動スキャンし、大規模モデルに株式のロングバスケットの画像を自動生成させ、AIサブセクターを自動構築し、BraveSearchを介してセクターと個別株の情報を自動取得し、セクター情報を自動要約し、各バスケットのリアルタイムの強弱を自動検出し、バスケット内の株式のフォロー関係を自動集計し、リーダー価格のブレイクアウト後にニュースフォローアップ調査を実施します。

ニュースによる説明が成立し、方向性が一致し、統計関係もサポートされる場合、システムは確認済みシグナル（confirmed signal）を生成します。デフォルトモードでは通知のみで実運用は行わず、tradeモードに切り替えた場合のみ取引します。

実運用ページではログだけでなく、システム概要、バスケットのリアルタイム状態、セクター情報、最近のシグナル、最近の取引、最近のステップを表形式で表示します。これにより、システムがどの段階で停止しているかを明確に把握できます：価格異常のみなのか、ニュースに説明がないのか、フォロワーが条件を満たさないのか、リスク管理がブロックしているのかです。

まだ成熟していない部分

正直に言うと、このシステムはまだ成熟した自動取引戦略とは言えません。

これはどちらかと言うとAIテーマローテーション研究フレームワークであり、「株式画像、セクター情報、価格異常、ニュースフォローアップ調査、統計検証、取引シグナル」を連携できるようになりましたが、さらに検証が必要です。

最も引き続き改善が必要なのはいくつかの部分です。

（1）株式画像の品質。大規模モデルが時々エッジの効いた株式をバスケットに含めることがあるため、asset_typeを強制し、ETF、指数、地域ファンドが混入するのを防ぐ必要があります。

（2）ニュースの品質。BraveSearchが取得したニュースは必ずしもコアニュースとは限らず、モデルは価格を本当に説明できるものを判断する必要があります。

（3）統計関係。過去のフォロー関係が将来も有効とは限らないため、継続的なレビューが必要です。

（4）取引リスク管理。自動トレードの前に、アカウントレベルのポジション、セクターエクスポージャー、日次損失、約定確認を追加する必要があります。

そのため、今はまずnotifyモードで実行し、confirmed signalの品質を確認するのが適しています。

しばらくしてシグナルが市場の異常な動きを確かに説明でき、かつフォロワーがその後も頻繁に反応することが確認できたら、少量ポジションでの実運用を検討します。

最後に

この戦略はAIに直接発注させるものではありません。

私は常に、大規模モデルが取引システムで最も適している役割は、「何を買うか教えて」ではなく、人間が継続的に維持することが難しい情報構造（企業の事業内容、セクター所属、リアルタイムニュース、イベントの解釈、資金のローテーションパス、潜在的な影響対象）を処理する手助けをすることだと考えています。

実際の取引は依然として価格、統計、リスク管理に任せるべきです。

AI時代の定量戦略は、単に大規模モデルのインターフェースを追加するだけではなく、リアルタイムデータ、テーマグラフ、ニュースイベント、価格構造、取引執行、レビュー説明を連鎖させるべきです。

このAI株式グラフ戦略は、その方向に沿った試みの一つです。

戦略ソースコード：AI産業チェーンTradFi契約グラフ取引システム