設計書: news_reminder 地政学リスク拡張

Issue #151 | 作成: 2026-03-27 | ステータス: Draft

1. 概要

既存の ops/news_reminder/ パイプラインに「地政学リスク」カテゴリを追加する。 AIニュース巡回（scanner.py）と同じ仕組みを活用し、地政学リスク系RSSソースの巡回・Haiku評価・Slack投稿を行う。 加えて、月次でBlackRock等のダッシュボードURLを取得しAI解釈を投稿する機能を追加する。

2. 既存パイプラインの拡張ポイント

3. 追加するRSSソース一覧

ソース選定の根拠

• primary（公的機関・シンクタンク）: 外務省・防衛省は一次情報源。CSISは米国の代表的安全保障シンクタンク。Reutersは速報性の高い通信社。信頼度が高く、IF層判断で起票可能
• secondary（メディア）: 日経・NHKは二次情報源。裏取りが必要な場合あり

4. sources.json スキーマ拡張

既存スキーマに変更なし。category フィールドに "geopolitical" を追加するだけ。

{
  "sources": [
    // 既存AIソース（category: "ai"）... そのまま

    // 追加: 地政学リスクソース
    {
      "name": "外務省プレスリリース",
      "url": "https://www.mofa.go.jp/mofaj/xml/rss20.xml",
      "max_articles": 10,
      "tier": "primary",
      "category": "geopolitical",
      "type": "rss"
    },
    {
      "name": "CSIS",
      "url": "https://www.csis.org/analysis/feed",
      "max_articles": 5,
      "tier": "primary",
      "category": "geopolitical",
      "type": "rss"
    }
    // ... 他のソースも同様
  ]
}

category の値で "ai" と "geopolitical" を区別する。 既存の _load_sources() はそのまま全ソースを返すが、cmd_scan 側でカテゴリフィルタする。

5. scanner.py の変更箇所

5.1 カテゴリ別評価プロンプト

現在 RELEVANCE_PROMPT / MNML_CONTEXT がAI専用にハードコードされている。 これをカテゴリ別に切り替える。

# scanner.py に追加

GEOPOLITICAL_CONTEXT = """\
合同会社MNML は以下の事業を行う一人法人:
- ITコンサル（CDP/DWH/Snowflake/マーケDX）
- 香水ブランド（原料輸入あり）

地政学リスク評価の観点:
1. 日本の対外関係・安全保障に影響するか
2. サプライチェーン（特に香料原料の輸入元）への影響があるか
3. 金融市場・為替への波及が見込まれるか
4. ITコンサル顧客企業（大手製造業・金融機関）の事業環境に影響するか
5. 経営判断に必要な中長期トレンドか
"""

GEOPOLITICAL_PROMPT = """\
以下のニュース記事を地政学リスクの観点から分析し、合同会社MNMLの経営判断に役立つか評価してください。

{context}

## 記事
タイトル: {title}
ソース: {source}
日付: {published}
概要: {summary}

## 評価基準
1. 関連度（high/medium/low）: MNMLの事業への影響度
2. リスク分類: military / sanctions / political / resources / technology / climate のいずれか
3. 影響範囲: direct（直接影響）/ indirect（間接影響）/ awareness（認識しておくべき）
4. 時間軸: immediate（即時）/ short_term（1-3ヶ月）/ medium_term（3-12ヶ月）/ long_term（1年超）

relevant=true は、MNMLの経営判断に具体的に有用な場合のみ。
一般的な国際ニュースや遠い地域の話題はfalse。

## 出力形式（JSONのみ出力）
{{"relevant": true, "relevance": "high", "risk_type": "sanctions", "impact_scope": "indirect", "timeframe": "short_term", "summary_ja": "日本語要約（2-3文）", "action": "経営上の推奨アクション", "reason": "推薦理由"}}
"""

5.2 プロンプト分岐ロジック

_evaluate_article() を修正し、記事の category に応じてプロンプトを切り替える。

# _evaluate_article の変更

def _evaluate_article(article: dict) -> dict | None:
    """Claude CLIで記事の関連度を評価する。カテゴリに応じてプロンプトを切り替え。"""
    category = article.get("category", "ai")

    if category == "geopolitical":
        context = GEOPOLITICAL_CONTEXT
        prompt_template = GEOPOLITICAL_PROMPT
    else:
        context = MNML_CONTEXT
        prompt_template = RELEVANCE_PROMPT

    prompt = prompt_template.format(
        context=context,
        title=article["title"],
        source=article["source"],
        published=article["published"],
        summary=article["summary"],
    )
    # 以下同じ（_call_claude → _extract_json）

5.3 cmd_scan に --category オプション追加

# cmd_scan のシグネチャ変更
def cmd_scan(
    *,
    dry_run: bool = False,
    json_output: bool = False,
    no_issue: bool = False,
    max_articles: int = 0,
    category: str = "",   # 追加: "ai", "geopolitical", ""（全カテゴリ）
) -> None:

    # ソース読み込み後にカテゴリフィルタ
    sources = _load_sources()
    if category:
        sources = [s for s in sources if s.get("category") == category]

5.4 Slack投稿のカテゴリ分離

_notify_slack() のスレッド管理を変更。カテゴリごとに別スレッドに投稿する。

# _notify_slack の変更
def _notify_slack(
    total_articles: int,
    relevant: list[tuple[dict, dict]],
    category: str = "ai",  # 追加
) -> None:
    # ...
    thread_ts = get_or_create_thread(client, settings.slack_channel, category)
    # ヘッダテキストもカテゴリに応じて変更

6. thread.py の変更

CATEGORY_HEADERS に地政学リスクカテゴリを追加する。

CATEGORY_HEADERS: dict[str, str] = {
    "ai": "🤖 AINews",
    "finance": "📊 金融",
    "geopolitical": "🌍 地政学リスク",  # 追加
}

7. 月次ダッシュボード解釈

7.1 方針

特定のURL（BlackRockダッシュボード等）を月初に取得し、Claudeに解釈させてSlackに投稿する。 Webページの取得には playwright（ops環境に導入済み）を使い、JS描画後のコンテンツを取得する。

7.2 新規モジュール: app/dashboard.py

# app/dashboard.py の設計

"""月次ダッシュボード解釈 — 外部URLを取得しAI分析結果をSlack投稿する。"""

# ダッシュボード定義（sources.json とは別管理。URLは固定で頻繁には変わらない）
DASHBOARDS: list[dict] = [
    {
        "name": "BlackRock Geopolitical Risk Dashboard",
        "url": "https://www.blackrock.com/corporate/insights/blackrock-investment-institute/interactive-charts/geopolitical-risk-dashboard",
        "description": "BlackRock Investment Institute の地政学リスクスコア",
    },
]

async def fetch_dashboard(url: str) -> str:
    """Playwrightでページを取得し、テキストコンテンツを抽出する。"""
    # playwright で JS 描画後のテキスト取得
    # タイムアウト: 30秒
    # 取得失敗時はエラーログのみ（ジョブ全体は止めない）
    ...

def interpret_dashboard(name: str, content: str) -> str:
    """Claude CLI でダッシュボード内容を解釈する。"""
    # プロンプト: ダッシュボード内容 → 日本語サマリー + リスクレベル変動 + 注目ポイント
    ...

def cmd_dashboard(*, dry_run: bool = False) -> None:
    """全ダッシュボードを取得・解釈・投稿する。"""
    # 1. 各ダッシュボードURLを取得
    # 2. Claude で解釈
    # 3. Slackの geopolitical スレッドに投稿
    ...

7.3 ダッシュボード解釈プロンプト

DASHBOARD_PROMPT = """\
以下は {name} の最新データです。

{content}

## タスク
1. 主要な地政学リスクスコアの現在値と前月比の変動を要約してください
2. 特に注目すべきリスク（上昇傾向のもの）を3つ以内で挙げてください
3. 日本企業（特にITコンサル・消費財）への影響を簡潔に分析してください

## 出力形式
- 要約は日本語で、Slack Block Kit の mrkdwn 形式で出力
- 箇条書きを活用し、簡潔に（全体500文字以内）
"""

8. scheduler.py の変更

8.1 日次スケジュール

既存の _run_daily() は scan --json --no-issue で全カテゴリをスキャンする。 変更方針: _run_daily() 内で2回スキャンを実行（AI + 地政学）し、それぞれの結果を別スレッドに投稿する。

# scheduler.py の _run_daily 変更案

async def _run_daily(self) -> None:
    # 既存: AIニューススキャン
    ai_result = await self._run_scanner(category="ai")
    if ai_result:
        await self._post_and_evaluate(ai_result, category="ai")

    # 追加: 地政学リスクスキャン
    geo_result = await self._run_scanner(category="geopolitical")
    if geo_result:
        await self._post_and_evaluate(geo_result, category="geopolitical")

async def _run_scanner(self, category: str = "") -> dict | None:
    """News Scanner をsubprocessで実行。--category オプションを追加。"""
    cmd = [
        str(PROJECT_ROOT / ".venv/bin/python"),
        "-m", "news_reminder.pipeline",
        "scan", "--json", "--no-issue", "--max-articles", "30",
    ]
    if category:
        cmd.extend(["--category", category])
    # 以下同じ

8.2 月次スケジュール（ダッシュボード解釈）

毎月1日に実行。_last_runs で月単位の重複実行を防止する。

# scheduler.py に追加

DASHBOARD_DAY = 1  # 毎月1日

# run() ループ内に追加
if now.day == DASHBOARD_DAY and self._last_runs.get("geo_dashboard") != today:
    self._last_runs["geo_dashboard"] = today
    self._save_last_runs()
    await self._run_dashboard()

async def _run_dashboard(self) -> None:
    """月次ダッシュボード解釈を subprocess で実行する。"""
    proc = await asyncio.create_subprocess_exec(
        str(PROJECT_ROOT / ".venv/bin/python"),
        "-m", "news_reminder.pipeline",
        "dashboard",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=str(PROJECT_ROOT),
    )
    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=300)
    if proc.returncode != 0:
        await self._notify_job_error("geo_dashboard", stderr.decode())

9. pipeline/cli.py の変更

# scan コマンドにオプション追加
scan_parser.add_argument(
    "--category", type=str, default="",
    help="カテゴリフィルタ（ai / geopolitical / 空=全て）"
)

# dashboard コマンド追加
dash_parser = sub.add_parser("dashboard", help="月次ダッシュボード解釈")
dash_parser.add_argument("--dry-run", action="store_true", help="確認のみ")

# main() にハンドラ追加
elif args.command == "dashboard":
    from news_reminder.app.dashboard import cmd_dashboard
    cmd_dashboard(dry_run=args.dry_run)

10. ファイル構成（新規・変更一覧）

11. データフロー

11.1 日次フロー（地政学リスク）

scheduler.py (7:00)
  ├─ _run_scanner(category="ai")        → 既存と同じ
  └─ _run_scanner(category="geopolitical")
       ↓
     scanner.py --category geopolitical --json --no-issue
       ├─ sources.json → category=geopolitical のソースのみ取得
       ├─ 各記事を GEOPOLITICAL_PROMPT で Haiku 評価
       └─ JSON結果を返却
       ↓
     scheduler.py
       ├─ get_or_create_thread("geopolitical") → 🌍 地政学リスク スレッド
       ├─ サマリー投稿（Block Kit）
       └─ IF層に判断依頼（skip / issue / consult）

11.2 月次フロー（ダッシュボード解釈）

scheduler.py (毎月1日)
  └─ _run_dashboard()
       ↓
     dashboard.py
       ├─ Playwright で BlackRock ダッシュボード取得
       ├─ テキスト抽出（or スクリーンショット + Vision）
       ├─ Claude CLI で日本語解釈
       └─ Slack "geopolitical" スレッドに投稿

12. Haiku評価プロンプトの出力フィールド比較

共通フィールド（relevant, relevance, summary_ja, action, reason）はスキャナのフィルタ・投稿ロジックで参照されるため、両カテゴリで必ず出力する。 カテゴリ固有フィールドは投稿テンプレートで表示分岐する。

13. Slack投稿フォーマット（地政学リスク）

🌍 *地政学リスク スキャン結果*
評価記事数: 15件 | 注目: 3件

────────────────────
**
_CSIS_

米国商務省が中国向け半導体装置の輸出規制対象を拡大。
日本の装置メーカーにも影響の可能性。

リスク分類: `technology` | 影響: `indirect` | 時間軸: `short_term`
*推奨アクション*: 顧客企業（製造業）への影響を注視

14. 前提条件・制約

• RSSフィードのURL有効性は実装時に疎通確認が必要（フィード停止・URL変更の可能性）
• BlackRockダッシュボードはSPAのため、Playwrightが必要。描画完了の待機が必要
• Haiku評価の追加コストは「記事数 × 1回」。地政学ソース6件 × max_articles で見積もり、1日あたり40-60記事程度の追加評価になる見込み
• 月次ダッシュボード解釈はPlaywright + Claude CLI（Sonnet推奨）で1回。コスト軽微
• playwright は ops/ 環境に導入済み（mail_filing で使用中）

15. トレードオフ・検討事項