2028年グローバル知能危機 — AIが引き起こす「豊穣のパラドックス」と金融システムへの波及

作成日: 2026-02-25 参考元: The 2028 Global Intelligence Crisis — Citrini Research & Alap Shah (2026-02-22) 関連記事: 05_AI_Agent_Abundance2035.md / 06_SolveEverything_Agent_Blueprint.md

目次

1. レポート概要：「AI強気派にとってのベアケース」
2. シナリオタイムライン：2026年→2028年
3. 3つのフィードバックループ
4. 摩擦ゼロ経済とAgent消費者の台頭
5. プライベートクレジットの連鎖崩壊
6. 住宅ローン危機 2.0
7. 政策対応の限界
8. 考察：Diamandisの「豊穣」とCitriniの「危機」の統合
9. AI Agentビジネスへの実践的示唆
10. 実装ガイド：危機を乗り越えるAgentシステム
11. 実装ガイド：摩擦ゼロAgent経済シミュレータ
12. 実装ガイド：労働市場モニタリングAgent
13. リスクシナリオ別ポジショニング
14. まとめ：AI Agent構築者への教訓

1. レポート概要：「AI強気派にとってのベアケース」

1.1 何が語られているのか

Citrini Researchのレポートは、2026年2月から2028年6月までのシナリオ分析として書かれたユニークな「未来からの金融メモ」である。核心的な問いは：

「もしAI強気派の見通しが正しく、知能が本当に安価なコモディティになったら…それは実は弱気材料ではないか？」

これはDiamandis/Wissner-Grossの「Solve Everything」（06_）が描く「豊穣社会」の暗い鏡像である。同じ技術的前提（AIの急速な進歩）から出発しながら、制度的・金融的な遅延（ラグ）に焦点を当てることで、全く異なる短期的結論に到達する。

1.2 基本データ（シナリオ内の数値）

1.3 これは予測ではなく「テールリスクのシナリオ分析」

著者は明確に「これは予測ではなくシナリオ」と述べている。しかし、各段階の因果連鎖は構造的に合理的であり、AI Agent構築者として無視できない内容を含んでいる。

2. シナリオタイムライン：2026年→2028年

Phase 1: SaaS崩壊（2025年後半–2026年）

ServiceNowの再帰的メカニズム:

ServiceNowの顧客が人員削減
  → ITサービス管理のシート数が減少
    → ServiceNowの売上が減少
      → ServiceNowが人員削減
        → ServiceNow自身がAI化を加速
          → さらにSaaS不要論が強化

Phase 2: 摩擦ゼロ経済の到来（2027年前半）

Phase 3: システミックリスクへの転化（2027年後半–2028年）

3. 3つのフィードバックループ

レポートの核心は、自然なブレーキのないネガティブ・フィードバックループの存在である。

ループ1: 実体経済ループ

AI能力向上
  → 企業が人件費を削減
    → 削減分をAI投資に再配分
      → さらにAI能力が向上
        → さらに人員削減
          → 消費者支出が減少
            → 需要低迷
              → 利益率低下
                → さらにAI投資で効率化
                  → (ループに戻る)

重要な非対称性: 上位10%の所得者が消費支出の50%超を占める。ホワイトカラー2%の雇用減少は、裁量的消費の3-4%減に相当する。

ループ2: 金融ループ

所得毀損
  → 住宅ローン延滞
    → 銀行損失
      → 信用引き締め
        → 資産効果の逆転
          → さらに消費減
            → (ループに戻る)

ループ3: 制度ループ

政策対応の遅れ
  → 不十分な対策
    → 社会不安（"Occupy Silicon Valley"）
      → 政治的膠着
        → さらに対応が遅れる
          → (ループに戻る)

「Ghost GDP」（幽霊GDP）の概念

GDPは国民経済計算上は成長するが、その成長は実体経済を循環しない。機械は裁量的消費をしないからだ。

これは「Solve Everything」が描くAbundance Flywheelの暗い鏡像。生産性向上が資本所有者に集中し、消費者には還流しない構造。

4. 摩擦ゼロ経済とAgent消費者の台頭

4.1 AI Agentによる仲介破壊

レポートで最もAI Agent構築者にとって重要な部分。消費者のAI Agentが既存ビジネスモデルの「摩擦」を体系的に解体するシナリオ：

4.2 「習慣的仲介」の終焉

「価格とフィットで最適化する機械は、あなたのお気に気に入りのアプリなど気にしない」

これはAI Agentビジネスにとって最大のチャンスであり最大の脅威でもある。

• チャンス: 消費者Agent市場は巨大。Agentが購買意思決定を代行する世界のインフラを構築する者が勝つ
• 脅威: 自社のAI Agent SaaSも、顧客のAgentによって常に比較・乗り換えされる対象になる

4.3 トークン消費の爆発

これはAI Agentインフラ（コンピュート、API、ベクトルDB）の需要が爆発的に増加することを意味する。

5. プライベートクレジットの連鎖崩壊

5.1 Zendesk債務不履行のケーススタディ

5.2 デイジーチェーン構造

米国生命保険会社
  → バミューダ/ケイマン再保険子会社
    → オフショアSPV群
      → プライベートクレジット（PE-backed SaaS債務）
        → ARR担保型ローン
          → SaaS企業の売上（← AI Agentが破壊中）

構造的問題: 各機関の資産は異なる形式で同一のリスク（ホワイトカラー生産性成長の継続）に賭けており、真の損失配分が不透明。

5.3 保険会社への波及

6. 住宅ローン危機 2.0

6.1 2008年との決定的な違い

6.2 隠れたストレス指標（2027年）

• HELOC（ホームエクイティローン）引き出し増加
• 401(k)（退職金）からの早期引き出し加速
• クレジットカード残高急増
• 住宅ローンは技術的に正常だが、貯蓄の枯渇で支払い中
• 債務対所得比率が倍増

6.3 地域別住宅価格下落（2028年6月）

テック/金融雇用比率40%超のZIPコードで、ファニーメイが早期延滞の上昇を報告。

7. 政策対応の限界

7.1 従来の政策ツールが効かない理由

7.2 財政のジレンマ

労働分配率低下（64%→46%）
  → 給与税収減
    → 所得税収減
      → 歳入12%不足
        → しかし社会保障需要は急増
          → 財政赤字拡大
            → 政策対応余地の縮小

7.3 「知能プレミアム」の巻き戻し

近代経済史の全期間にわたり、人間の認知能力は希少であり、プレミアム報酬を得てきた。AIはこの希少な入力を広範に代替する初めての技術である。すべての制度——労働市場、住宅ローン、税制——は知能の希少性に最適化されていた。再価格付けは無秩序なものになる。

8. 考察：Diamandisの「豊穣」とCitriniの「危機」の統合

8.1 同じコインの表と裏

3つの記事を並べると、同じ技術的前提から異なる時間軸と焦点で全く異なる結論に至ることが分かる：

8.2 統合フレームワーク：「豊穣のパラドックス」

長期的には豊穣（Diamandis）は正しい
  BUT
短期的には移行コスト（Citrini）が壊滅的になりうる
  BECAUSE
制度（ローン、税制、保険、雇用）の再設計速度 << AI能力の進化速度

これが「豊穣のパラドックス」:
  技術的には十分に豊穣なのに、制度的ボトルネックが
  その恩恵を分配できず、むしろ危機を引き起こす

8.3 Solve Everythingの「The Muddle」vs Citriniの「3つのループ」

Diamandisが「The Muddle（停滞・官僚機構）」と呼んだものの具体的な金融メカニズムをCitriniが描いている：

8.4 AI Agent構築者にとっての意味

Diamandisだけを読むと: 評価基盤を構築すれば勝てる → 技術に集中

Citriniも読むと: 移行期のリスクを理解し、自社の顧客基盤が構造的に毀損されるリスクを計算に入れる必要がある

• ホワイトカラー向けAgent SaaS → 顧客自体が消滅するリスク
• 消費者向けAgent → 消費支出自体が縮小するリスク
• 金融向けAgent → 金融システム自体が揺らぐリスク

結論: 短期的な収益モデルは構造的労働置換に対して脆弱な層を避け、移行期を支えるインフラ（再訓練Agent、リスク監視Agent、政策シミュレーションAgent）に軸足を置くべき。

9. AI Agentビジネスへの実践的示唆

9.1 脆弱セクター（避けるべき顧客層）

9.2 強靭セクター（注力すべき領域）

9.3 収益モデルの耐性チェック

問い: 自社のAI Agent SaaSの顧客が30%の人員を削減したら、
     売上はどうなるか？

パターンA: シートベース課金 → 売上30%減（ServiceNowパターン）
パターンB: 成果ベース課金  → 成果が出ている限り維持
パターンC: インフラ課金    → トークン消費量に比例して増加の可能性

→ パターンCが最も耐性が高い
→ Solve Everythingの「Outcome Procurement」はパターンBに対応

10. 実装ガイド：危機を乗り越えるAgentシステム

10.1 Intelligence Premium Monitor（知能プレミアム監視Agent）

"""
Intelligence Premium Monitor
「知能プレミアムの巻き戻し」をリアルタイム監視する。
Citriniが指摘する「知能の希少性→コモディティ化」の進行度を計測。
"""

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class AlertLevel(Enum):
    GREEN = "normal"
    YELLOW = "caution"
    ORANGE = "elevated"
    RED = "critical"


@dataclass
class LaborMarketSignal:
    """労働市場シグナル"""
    metric: str
    value: float
    previous_value: float
    source: str
    timestamp: datetime
    sector: str  # "white_collar" | "blue_collar" | "overall"

    @property
    def change_pct(self) -> float:
        if self.previous_value == 0:
            return 0
        return (self.value - self.previous_value) / self.previous_value * 100


@dataclass
class FrictionCollapseSignal:
    """摩擦崩壊シグナル（セクター別）"""
    sector: str
    metric: str            # "commission_rate", "margin", "churn_rate"
    current_value: float
    baseline_value: float  # 崩壊前の基準値
    collapse_ratio: float  # 崩壊の進行度（0.0–1.0）


@dataclass
class CrisisAssessment:
    """危機評価レポート"""
    timestamp: datetime
    alert_level: AlertLevel
    labor_stress: float          # 0.0–1.0
    financial_stress: float      # 0.0–1.0
    friction_collapse: float     # 0.0–1.0
    ghost_gdp_indicator: float   # 生産性成長 vs 消費成長の乖離
    narrative: str
    recommended_actions: list[str]


class IntelligencePremiumMonitor:
    """
    知能プレミアムの巻き戻しを監視するAgent。

    Citriniの3つのフィードバックループを追跡:
    1. 実体経済ループ（労働市場→消費→利益率）
    2. 金融ループ（所得→ローン→信用）
    3. 制度ループ（政策対応→社会不安）
    """

    # Citriniが示すスレッショルド
    THRESHOLDS = {
        "jolts_yoy_decline": -0.15,          # JOLTS求人 15%YoY減
        "initial_claims": 400_000,            # 失業保険申請 40万件超
        "white_collar_posting_decline": -0.20, # WC求人 20%減
        "consumer_spending_decline": -0.03,    # 消費支出 3%減
        "card_volume_growth_slowdown": 0.035,  # カード決済成長率 3.5%以下
        "saas_revenue_deceleration": -0.30,    # SaaS収益成長30%減速
    }

    def __init__(self):
        self.labor_signals: list[LaborMarketSignal] = []
        self.friction_signals: list[FrictionCollapseSignal] = []

    def ingest_labor_signal(self, signal: LaborMarketSignal) -> None:
        self.labor_signals.append(signal)

    def ingest_friction_signal(self, signal: FrictionCollapseSignal) -> None:
        self.friction_signals.append(signal)

    def assess(self) -> CrisisAssessment:
        """総合的な危機評価"""
        labor = self._assess_labor_stress()
        financial = self._assess_financial_stress()
        friction = self._assess_friction_collapse()
        ghost_gdp = self._assess_ghost_gdp()

        # 総合アラートレベル
        max_stress = max(labor, financial, friction)
        if max_stress >= 0.8:
            level = AlertLevel.RED
        elif max_stress >= 0.6:
            level = AlertLevel.ORANGE
        elif max_stress >= 0.4:
            level = AlertLevel.YELLOW
        else:
            level = AlertLevel.GREEN

        return CrisisAssessment(
            timestamp=datetime.now(),
            alert_level=level,
            labor_stress=labor,
            financial_stress=financial,
            friction_collapse=friction,
            ghost_gdp_indicator=ghost_gdp,
            narrative=self._generate_narrative(level, labor, financial, friction),
            recommended_actions=self._recommend_actions(level),
        )

    def _assess_labor_stress(self) -> float:
        """労働市場ストレスの評価（0.0–1.0）"""
        if not self.labor_signals:
            return 0.0

        wc_signals = [s for s in self.labor_signals if s.sector == "white_collar"]
        if not wc_signals:
            return 0.0

        # ホワイトカラー求人の変化率で評価
        latest = max(wc_signals, key=lambda s: s.timestamp)
        decline = abs(min(latest.change_pct / 100, 0))
        return min(decline / 0.30, 1.0)  # 30%減少で1.0

    def _assess_financial_stress(self) -> float:
        """金融ストレスの評価"""
        # 実装では住宅価格、HELOC引き出し、401k早期引出し等を追跡
        return 0.0

    def _assess_friction_collapse(self) -> float:
        """摩擦崩壊の評価"""
        if not self.friction_signals:
            return 0.0
        avg = sum(s.collapse_ratio for s in self.friction_signals) / len(self.friction_signals)
        return min(avg, 1.0)

    def _assess_ghost_gdp(self) -> float:
        """Ghost GDP指標: 生産性成長と消費成長の乖離"""
        # 生産性成長 - 消費成長 が大きいほどGhost GDP
        return 0.0

    def _generate_narrative(
        self, level: AlertLevel, labor: float, financial: float, friction: float
    ) -> str:
        if level == AlertLevel.RED:
            return (
                "⚠️ 知能プレミアムの急速な巻き戻しが進行中。"
                "ホワイトカラー労働市場の構造的毀損が金融システムに波及するリスク。"
                "Citriniシナリオのフェーズ2-3に該当する兆候。"
            )
        elif level == AlertLevel.ORANGE:
            return (
                "🟠 複数のストレス指標が上昇。"
                f"労働: {labor:.0%}, 金融: {financial:.0%}, 摩擦崩壊: {friction:.0%}。"
                "ServiceNowパターンの再帰的崩壊に注意。"
            )
        elif level == AlertLevel.YELLOW:
            return (
                "🟡 初期兆候を検出。"
                "特定セクターでのAgent化が加速。摩擦ベースの収益モデルに注意。"
            )
        return "🟢 現時点で顕著なストレスなし。通常監視を継続。"

    def _recommend_actions(self, level: AlertLevel) -> list[str]:
        base = ["ホワイトカラー求人データの週次モニタリング継続"]
        if level.value in ("critical", "elevated"):
            return base + [
                "シートベース課金→成果ベース課金への移行を加速",
                "ホワイトカラー集中顧客へのエクスポージャーを削減",
                "ブルーカラー/物理世界Agentへのピボットを検討",
                "キャッシュポジションの強化（6ヶ月分のランウェイ確保）",
                "Ghost GDP指標の導入（生産性vs消費の乖離追跡）",
            ]
        elif level.value == "caution":
            return base + [
                "収益モデルの耐性テスト（顧客30%人員削減シナリオ）",
                "成果ベース課金のパイロット開始",
            ]
        return base

10.2 FastAPIエンドポイント

"""
Intelligence Premium Monitor API
"""

from fastapi import APIRouter
from pydantic import BaseModel
from datetime import datetime

router = APIRouter(prefix="/crisis-monitor", tags=["Crisis Monitor"])


class LaborSignalInput(BaseModel):
    metric: str
    value: float
    previous_value: float
    source: str
    sector: str


class FrictionSignalInput(BaseModel):
    sector: str
    metric: str
    current_value: float
    baseline_value: float
    collapse_ratio: float


class CrisisReport(BaseModel):
    alert_level: str
    labor_stress: float
    financial_stress: float
    friction_collapse: float
    ghost_gdp_indicator: float
    narrative: str
    recommended_actions: list[str]


# グローバルインスタンス
monitor = IntelligencePremiumMonitor()


@router.post("/signals/labor")
async def submit_labor_signal(signal: LaborSignalInput):
    monitor.ingest_labor_signal(LaborMarketSignal(
        timestamp=datetime.now(), **signal.model_dump(),
    ))
    return {"status": "accepted"}


@router.post("/signals/friction")
async def submit_friction_signal(signal: FrictionSignalInput):
    monitor.ingest_friction_signal(FrictionCollapseSignal(**signal.model_dump()))
    return {"status": "accepted"}


@router.get("/assessment", response_model=CrisisReport)
async def get_crisis_assessment():
    result = monitor.assess()
    return CrisisReport(
        alert_level=result.alert_level.value,
        labor_stress=result.labor_stress,
        financial_stress=result.financial_stress,
        friction_collapse=result.friction_collapse,
        ghost_gdp_indicator=result.ghost_gdp_indicator,
        narrative=result.narrative,
        recommended_actions=result.recommended_actions,
    )

11. 実装ガイド：摩擦ゼロAgent経済シミュレータ

Citriniが描く「Agent消費者がビジネスモデルの摩擦を破壊する」プロセスをシミュレーションするシステム：

"""
Friction-Zero Economy Simulator
AI Agentが各セクターの「摩擦」を解体するプロセスをモデル化。
自社ビジネスモデルの耐性をテストする。
"""

from dataclasses import dataclass, field
from typing import Any


@dataclass
class BusinessModel:
    """ビジネスモデル定義"""
    name: str
    sector: str
    annual_revenue_usd: float
    margin_pct: float
    revenue_sources: list[dict[str, Any]]  # {name, amount, friction_type}
    customer_segments: list[dict[str, Any]]  # {name, pct, white_collar_pct}


@dataclass
class FrictionType:
    """摩擦の種類と脆弱性"""
    name: str
    description: str
    agent_vulnerability: float  # 0.0–1.0（Agentによる破壊されやすさ）
    time_to_collapse_months: int


# Citriniが特定した摩擦タイプ
FRICTION_TYPES = {
    "switching_cost": FrictionType(
        "スイッチングコスト", "乗り換えの手間・学習コスト", 0.9, 12
    ),
    "information_asymmetry": FrictionType(
        "情報の非対称性", "専門知識の壁", 0.85, 18
    ),
    "habitual_usage": FrictionType(
        "習慣的利用", "アプリ忠誠度、惰性的更新", 0.95, 6
    ),
    "comparison_cost": FrictionType(
        "比較コスト", "複数サービスの比較にかかる時間", 0.95, 6
    ),
    "regulatory_moat": FrictionType(
        "規制による堀", "免許・認可の壁", 0.3, 36
    ),
    "physical_presence": FrictionType(
        "物理的存在", "対面・現場が必要", 0.2, 48
    ),
    "network_effect": FrictionType(
        "ネットワーク効果", "ユーザー数がサービス価値を決定", 0.5, 24
    ),
}


@dataclass
class SimulationResult:
    """シミュレーション結果"""
    business_name: str
    scenario: str
    original_revenue: float
    projected_revenue: float
    revenue_decline_pct: float
    most_vulnerable_source: str
    survival_probability: float
    recommendations: list[str]


class FrictionZeroSimulator:
    """
    摩擦ゼロ経済シミュレータ。
    自社ビジネスモデルの各収益源が
    AI Agentによってどの程度破壊されるかをモデル化する。
    """

    def simulate(
        self, model: BusinessModel, agent_penetration: float = 0.5
    ) -> SimulationResult:
        """
        agent_penetration: 消費者のうちAI Agentを利用する割合（0.0–1.0）
        Citriniシナリオでは2027年3月時点で高い浸透率に到達
        """
        total_original = model.annual_revenue_usd
        total_projected = 0.0
        most_vulnerable = ""
        max_decline = 0.0

        for source in model.revenue_sources:
            friction = FRICTION_TYPES.get(source.get("friction_type", ""))
            if not friction:
                total_projected += source["amount"]
                continue

            # Agent浸透率 × 脆弱性 = 収益減少率
            decline_rate = agent_penetration * friction.agent_vulnerability
            projected = source["amount"] * (1 - decline_rate)
            total_projected += projected

            if decline_rate > max_decline:
                max_decline = decline_rate
                most_vulnerable = source["name"]

        # ホワイトカラー顧客の構造的リスク（Citriniの消費乗数効果）
        wc_exposure = sum(
            seg["pct"] * seg.get("white_collar_pct", 0)
            for seg in model.customer_segments
        )
        # 2%のWC雇用減 = 3-4%の裁量的消費減（Citriniの乗数）
        wc_multiplier = 1 - (wc_exposure * 0.02 * 1.75)  # 中央値の乗数
        total_projected *= max(wc_multiplier, 0.5)

        decline_pct = (1 - total_projected / total_original) * 100 if total_original > 0 else 0
        survival = max(0, 1 - decline_pct / 100)

        return SimulationResult(
            business_name=model.name,
            scenario=f"Agent浸透率{agent_penetration:.0%}",
            original_revenue=total_original,
            projected_revenue=round(total_projected, 2),
            revenue_decline_pct=round(decline_pct, 1),
            most_vulnerable_source=most_vulnerable,
            survival_probability=round(survival, 2),
            recommendations=self._generate_recs(decline_pct, most_vulnerable),
        )

    def stress_test(
        self, model: BusinessModel, scenarios: list[float] = None
    ) -> list[SimulationResult]:
        """複数のAgent浸透率でストレステスト"""
        if scenarios is None:
            scenarios = [0.1, 0.3, 0.5, 0.7, 0.9]
        return [self.simulate(model, pen) for pen in scenarios]

    def _generate_recs(self, decline: float, vulnerable: str) -> list[str]:
        recs = []
        if decline > 50:
            recs.append(f"⚠️ 致命的リスク: '{vulnerable}'の収益モデルを根本的に再設計")
            recs.append("成果ベース課金への即時移行を推奨")
            recs.append("ブルーカラー/物理世界の顧客基盤への多角化")
        elif decline > 30:
            recs.append(f"🟠 高リスク: '{vulnerable}'への依存度を削減")
            recs.append("6ヶ月以内に代替収益源を構築")
        elif decline > 15:
            recs.append(f"🟡 要注意: '{vulnerable}'の摩擦が崩壊する可能性")
            recs.append("Agent耐性のある差別化要素を強化")
        return recs


# --- 使用例: AI Agent SaaSビジネスのストレステスト ---

simulator = FrictionZeroSimulator()

my_saas = BusinessModel(
    name="AI Agent Lab SaaS",
    sector="ai_agent_saas",
    annual_revenue_usd=5_000_000,
    margin_pct=70,
    revenue_sources=[
        {"name": "月額サブスク（シートベース）", "amount": 3_000_000, "friction_type": "switching_cost"},
        {"name": "API従量課金", "amount": 1_200_000, "friction_type": "network_effect"},
        {"name": "コンサルティング", "amount": 800_000, "friction_type": "information_asymmetry"},
    ],
    customer_segments=[
        {"name": "テック企業", "pct": 0.4, "white_collar_pct": 0.85},
        {"name": "金融機関", "pct": 0.3, "white_collar_pct": 0.75},
        {"name": "製造業", "pct": 0.2, "white_collar_pct": 0.40},
        {"name": "公共/教育", "pct": 0.1, "white_collar_pct": 0.60},
    ],
)

results = simulator.stress_test(my_saas)
for r in results:
    print(f"{r.scenario}: 売上 ${r.original_revenue:,.0f} → ${r.projected_revenue:,.0f} "
          f"({r.revenue_decline_pct:+.1f}%) | 生存確率: {r.survival_probability:.0%}")

12. 実装ガイド：労働市場モニタリングAgent

"""
Labor Market Transition Monitor
Citriniが指摘する「ホワイトカラー vs ブルーカラー」の
二極化をリアルタイム監視するAgent。
"""

from dataclasses import dataclass
from datetime import datetime
from typing import Any

from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter(prefix="/labor-monitor", tags=["Labor Market"])


@dataclass
class SectorHealth:
    sector: str
    job_postings_change_pct: float   # YoY求人変化率
    wage_growth_pct: float           # 実質賃金成長率
    ai_adoption_rate: float          # AI導入率
    displacement_risk: float         # 0.0–1.0
    citrini_phase: str               # "pre-crisis" | "phase1" | "phase2" | "phase3"


class LaborTransitionMonitor:
    """
    Citriniタイムラインに基づく労働市場監視。

    Phase 1 (2025後半-2026): SaaS崩壊、Agentic Coding成熟
    Phase 2 (2027前半): 摩擦ゼロ経済、トークン消費爆発
    Phase 3 (2027後半-2028): システミックリスク、住宅ローン危機
    """

    def __init__(self):
        self.sector_data: dict[str, SectorHealth] = {}

    def update_sector(self, health: SectorHealth) -> None:
        self.sector_data[health.sector] = health

    def detect_citrini_phase(self) -> dict[str, Any]:
        """Citriniシナリオのどのフェーズにいるかを判定"""
        wc_sectors = {k: v for k, v in self.sector_data.items()
                      if k in ("software", "finance", "consulting", "legal")}
        bc_sectors = {k: v for k, v in self.sector_data.items()
                      if k in ("manufacturing", "logistics", "healthcare", "construction")}

        if not wc_sectors:
            return {"phase": "unknown", "confidence": 0}

        wc_avg_posting = sum(s.job_postings_change_pct for s in wc_sectors.values()) / len(wc_sectors)
        wc_avg_wage = sum(s.wage_growth_pct for s in wc_sectors.values()) / len(wc_sectors)

        bc_avg_posting = 0.0
        if bc_sectors:
            bc_avg_posting = sum(s.job_postings_change_pct for s in bc_sectors.values()) / len(bc_sectors)

        # Citriniの兆候: WC求人崩壊 + BC安定 = 二極化
        divergence = bc_avg_posting - wc_avg_posting

        if wc_avg_posting < -20 and divergence > 15:
            phase = "phase2_friction_zero"
            desc = "ホワイトカラー求人が崩壊しブルーカラーは安定。Citrini Phase 2に該当"
        elif wc_avg_posting < -10:
            phase = "phase1_saas_collapse"
            desc = "ホワイトカラー求人が減少開始。Citrini Phase 1の兆候"
        elif wc_avg_posting < -5 and wc_avg_wage < 0:
            phase = "pre_crisis_warning"
            desc = "求人減速と実質賃金のマイナス成長が同時進行"
        else:
            phase = "normal"
            desc = "顕著な兆候なし"

        return {
            "phase": phase,
            "description": desc,
            "white_collar_posting_change": f"{wc_avg_posting:+.1f}%",
            "white_collar_wage_growth": f"{wc_avg_wage:+.1f}%",
            "blue_collar_posting_change": f"{bc_avg_posting:+.1f}%",
            "wc_bc_divergence": f"{divergence:+.1f}pp",
            "serviceNow_pattern_risk": wc_avg_posting < -15,
        }

    def consumption_impact_estimate(self) -> dict[str, Any]:
        """
        消費への波及効果を推計。
        Citriniの乗数: WC雇用2%減 = 裁量的消費3-4%減
        """
        wc = [v for k, v in self.sector_data.items()
              if k in ("software", "finance", "consulting", "legal")]
        if not wc:
            return {"estimate": "data_insufficient"}

        avg_displacement = sum(s.displacement_risk for s in wc) / len(wc)

        # 上位10%所得者が消費の50%超を占める（Citrini）
        estimated_wc_employment_decline = avg_displacement * 0.10  # 最大10%雇用減
        consumption_impact = estimated_wc_employment_decline * 1.75  # 乗数1.75

        return {
            "wc_displacement_risk": f"{avg_displacement:.0%}",
            "estimated_employment_decline": f"{estimated_wc_employment_decline:.1%}",
            "estimated_consumption_impact": f"-{consumption_impact:.1%}",
            "ghost_gdp_risk": "HIGH" if consumption_impact > 0.05 else "MODERATE",
            "note": "上位10%所得者が消費の50%超。WC雇用減の消費インパクトは非対称的に大きい",
        }


# --- APIエンドポイント ---

class SectorInput(BaseModel):
    sector: str
    job_postings_change_pct: float
    wage_growth_pct: float
    ai_adoption_rate: float
    displacement_risk: float


labor_monitor = LaborTransitionMonitor()


@router.post("/sectors")
async def update_sector(data: SectorInput):
    labor_monitor.update_sector(SectorHealth(
        **data.model_dump(), citrini_phase="",
    ))
    return {"status": "updated"}


@router.get("/phase-detection")
async def detect_phase():
    return labor_monitor.detect_citrini_phase()


@router.get("/consumption-impact")
async def consumption_impact():
    return labor_monitor.consumption_impact_estimate()

13. リスクシナリオ別ポジショニング

13.1 3シナリオ比較

13.2 シナリオ横断で有効な戦略

全シナリオで堅実な投資先は：

1. AI評価基盤（ベンチマーク） — 市場が拡大しても縮小しても、評価の需要は増加
2. 成果ベース課金（Outcome Procurement） — 顧客人員削減に耐性
3. ブルーカラー/物理世界Agent — Citriniもブルーカラーは安定と指摘
4. リスク監視Agent — 危機時こそ需要増大
5. キャッシュポジション — 全シナリオで6ヶ月以上のランウェイ

14. まとめ：AI Agent構築者への教訓

Citriniレポートの核心メッセージ

「カナリアはまだ生きている」 — しかし、坑道の空気は変わり始めている。

3つの記事を統合した戦略フレームワーク

「豊穣のパラドックス」を乗り越える設計原則

1. シートベース課金を捨てよ — 顧客が人員削減しても売上が維持される収益モデルに
2. Ghost GDPを監視せよ — 生産性成長と消費成長の乖離が危機の先行指標
3. 摩擦を堀にするな — AI Agentが破壊する「摩擦」に依存したビジネスモデルは崩壊する
4. 移行期を支えるAgentを作れ — 再訓練、リスク監視、政策シミュレーションは全シナリオで需要
5. 物理世界に根を張れ — ブルーカラー/物理世界のAgentはCitriniシナリオでも安定

参考文献

• Citrini Research & Shah, A. (2026). "The 2028 Global Intelligence Crisis". Citrini Research, February 22, 2026.
• Wissner-Gross, A. D. & Diamandis, P. H. (2026). "Solve Everything".
• Diamandis, P. H. (2026). "How We Get to Abundance by 2035". Metatrends.
• 関連記事: 05_AI_Agent_Abundance2035.md
• 関連記事: 06_SolveEverything_Agent_Blueprint.md