news_reminder パイプライン再設計

Issue: #328 | 作成日: 2026-04-07 | 種別: アーキテクチャ設計書

1. 現状の課題

2. 設計コンセプト

設計の柱

1. ドメイン = JSON設定ファイル — ドメイン追加はJSONファイル1つ追加で完結
2. ソース = アダプターパターン — RSS / HN API / arXiv API / Google News RSS を統一インターフェースで扱う。将来のX API追加もアダプター1つ
3. パイプライン = 4ステップ固定 — 取得 → フィルタ → 要約 → 投稿
4. LLM呼び出し = 1モジュールに集約 — Claude CLIのラッパーを統一
5. Slack投稿 = ドメイン別テンプレート — 金融は銘柄影響表、AIはMNML改修余地など

3. ドメイン定義

3.1 ドメイン設定ファイル

各ドメインを data/domains/{ドメイン名}.json で定義する。ファイル名はドメイン名をそのまま使う（例: ai.json, finance.json, geopolitics.json）。

AI ドメイン — data/domains/ai.json

{
  "id": "ai",
  "name": "AI",
  "emoji": "🤖",
  "thread_prefix": "AINews",
  "sources": ["openai_blog", "google_ai_blog", "deepmind_blog",
              "meta_engineering", "ms_research", "huggingface_blog",
              "arxiv_ai", "techcrunch_ai", "theverge_ai",
              "mit_tech_review", "hn_ai"],
  "evaluate_prompt": "prompts/evaluate_ai.txt",
  "summarize_prompt": "prompts/summarize_ai.txt",
  "formatter": "ai",
  "max_articles_per_run": 30,
  "mention_ceo": true
}

金融ドメイン — data/domains/finance.json

{
  "id": "finance",
  "name": "金融",
  "emoji": "📊",
  "thread_prefix": "金融News",
  "sources": ["google_news"],
  "source_config": {
    "google_news": {
      "holdings_file": "data/holdings.json"
    }
  },
  "evaluate_prompt": "prompts/evaluate_finance.txt",
  "summarize_prompt": "prompts/summarize_finance.txt",
  "formatter": "finance",
  "max_articles_per_run": 50,
  "mention_ceo": true
}

地政学ドメイン — data/domains/geopolitics.json

{
  "id": "geopolitics",
  "name": "地政学",
  "emoji": "🌍",
  "thread_prefix": "地政学News",
  "sources": ["bbc_world", "aljazeera", "nhk_intl", "nhk_politics", "jiji"],
  "evaluate_prompt": "prompts/evaluate_geopolitics.txt",
  "summarize_prompt": "prompts/summarize_geopolitics.txt",
  "formatter": "default",
  "max_articles_per_run": 30,
  "mention_ceo": true
}

3.2 ドメイン設定のスキーマ

3.3 ドメイン追加の手順

4. ソース抽象化

4.1 ソース定義ファイル

data/sources.json を継続利用。現行フォーマットをベースに id フィールドを追加する。

{
  "sources": [
    {
      "id": "openai_blog",
      "name": "OpenAI Blog",
      "type": "rss",
      "url": "https://openai.com/blog/rss.xml",
      "max_articles": 5,
      "tier": "primary"
    },
    {
      "id": "arxiv_ai",
      "name": "arXiv AI",
      "type": "arxiv_api",
      "categories": ["cs.AI", "cs.CL", "cs.LG"],
      "max_articles": 20,
      "tier": "primary"
    },
    {
      "id": "hn_ai",
      "name": "Hacker News AI",
      "type": "hn_api",
      "min_score": 50,
      "max_articles": 15,
      "tier": "secondary"
    },
    {
      "id": "google_news",
      "name": "Google News (銘柄検索)",
      "type": "google_news_rss",
      "max_articles": 20,
      "tier": "primary"
    }
  ]
}

4.2 ソースアダプター

各ソースタイプに対応するアダプタークラスを app/sources/ に配置する。

4.3 共通インターフェース

class SourceAdapter(Protocol):
    """ソースアダプターの共通インターフェース"""

    async def fetch(
        self,
        source_config: dict,
        domain_config: DomainConfig,
    ) -> list[Article]:
        """記事を取得して返す。"""
        ...

@dataclass
class Article:
    """取得した記事の統一データモデル"""
    id: str               # URL等から生成したハッシュ
    title: str
    url: str
    source_name: str      # ソース名（例: "OpenAI Blog"）
    source_id: str        # ソースID（例: "openai_blog"）
    published: datetime | None
    content: str          # 本文（あれば。RSSのsummary等）
    tier: str             # "primary" | "secondary"
    metadata: dict        # ソース固有の追加情報（HNのscore、arXivの著者等）

google_news_rss タイプは domain_config.source_config から holdings_file パスを受け取り、銘柄ごとの検索語でRSSを取得する。返す Article の metadata に ticker / holding_name を含める。

5. 共通パイプライン

5.1 パイプライン4ステップ

5.2 パイプラインのエントリーポイント

async def run_domain_pipeline(
    domain_id: str,
    *,
    slack_client: AsyncWebClient | None = None,
    dry_run: bool = False,
) -> PipelineResult:
    """1ドメインのパイプラインを実行する。

    Args:
        domain_id: ドメインID（例: "ai", "finance", "geopolitics"）
        slack_client: Slack APIクライアント（None時はCLI実行で投稿スキップ）
        dry_run: True時はLLM呼び出し・Slack投稿をスキップ

    Returns:
        PipelineResult: 実行結果（記事数、候補数等）
    """
    ...

5.3 バッチ評価によるスループット向上

評価プロンプトに複数記事をまとめて渡し、JSON配列で結果を返させる。バッチサイズはデフォルト20件。Haikuのコンテキスト長（200K）に対して十分な余裕がある。設定で変更可能。

# バッチ評価の処理フロー
articles = fetch_step(domain)          # 例: 30件
batches = chunk(articles, size=20)     # 2バッチに分割
for batch in batches:
    evaluations = await llm.evaluate_batch(batch, domain.evaluate_prompt)
    for article, eval in zip(batch, evaluations):
        article.evaluation = eval

バッチサイズは Settings.evaluate_batch_size で設定可能（デフォルト: 20）。

6. LLM呼び出し統一

6.1 統一モジュール app/llm.py

現行3種のClaude CLI呼び出しを1モジュールに統合する。

class ClaudeCli:
    """Claude CLI統一ラッパー"""

    def __init__(
        self,
        cli_path: str = "claude",
        default_model: str = "haiku",
        timeout: int = 120,
    ):
        ...

    async def run(
        self,
        prompt: str,
        *,
        model: str | None = None,    # None時はdefault_model
        timeout: int | None = None,   # None時はself.timeout
        json_output: bool = False,    # True時はJSON抽出を試みる
    ) -> str:
        """Claude CLIを実行しテキスト結果を返す。"""
        ...

    async def run_json(
        self,
        prompt: str,
        *,
        model: str | None = None,
        timeout: int | None = None,
    ) -> dict | list:
        """Claude CLIを実行しJSONをパースして返す。"""
        ...

6.2 統一する設定

7. Slack投稿設計

7.1 スレッド管理

現行の thread.py を拡張する。CEO要件の「20260407_ドメイン_News」形式に対応。

7.2 投稿構成（ドメイン別フォーマッタ）

共通フォーマッタ（default）— 地政学等

1. サマリー（全体の要約、1ブロック）
2. 重要ニュース（記事ごとのセクション）
3. リファレンス（URL一覧）
4. CEOメンション

AIフォーマッタ（ai）

1. サマリー
2. MNML-agent改修余地（該当記事がある場合のみ。LLM要約で抽出）
3. 重要ニュース
4. リファレンス
5. CEOメンション

金融フォーマッタ（finance）

1. サマリー（市場全体の概況）
2. 保有銘柄ごとの影響・アクション（表形式）
3. 重要ニュース
4. リファレンス
5. CEOメンション

7.3 金融ドメインの銘柄影響表（Block Kit）

Slack Block Kitでは <table> が使えないため、Markdown風テーブルで表現する。

📊 *保有銘柄影響*

| 銘柄 | 影響 | アクション |
|------|------|-----------|
| 🟢 eMAXIS S&P500 | ポジティブ: FRB利下げ示唆 | 静観（積立継続） |
| 🔴 リクルート | ネガティブ: 求人広告規制強化 | 決算注視 |
| ⚪ BTC | 中立: ETF資金流入鈍化 | 静観 |

mrkdwn 型の section ブロックで投稿する。3000文字制限があるため、銘柄数が多い場合は複数ブロックに分割する。

7.4 CEOメンション

各投稿の末尾に <@U0AHXTRDQMA> を付与する。mention_ceo: true のドメインのみ。

8. スケジューラ統合

8.1 統合先: bot/scheduler.py の DailyScheduler

DailyScheduler の現行ジョブ一覧

8.2 新ジョブ: news_domain

8.3 scheduler.py への統合

現行の news_scan（07:00）と news_pipeline（08:00/12:00/18:00）を廃止し、news_domain ジョブに一本化する。

# bot/scheduler.py 変更箇所

# 定数
NEWS_DOMAIN_HOURS = [7, 12, 18]  # 旧: NEWS_PIPELINE_HOURS + scan_hour

# run() ループ内
if now.hour in NEWS_DOMAIN_HOURS:
    key = f"news_domain_{now.hour:02d}"
    if self._last_runs.get(key) != today:
        self._last_runs[key] = today
        self._save_last_runs()
        await self._run_all_domains()

async def _run_all_domains(self) -> None:
    """全ドメインのパイプラインを順次実行する。"""
    domain_ids = load_all_domain_ids()  # data/domains/*.json を走査
    for domain_id in domain_ids:
        try:
            await run_domain_pipeline(
                domain_id,
                slack_client=self.client,
            )
        except Exception:
            log.exception("ドメイン %s のパイプラインでエラー", domain_id)
            await self._notify_job_error(f"news_{domain_id}", ...)

8.4 廃止するジョブ・プロセス

9. データ管理・ローテーション

9.1 scanned.json ローテーション

既読記事の重複判定用。現行は1,000件上限のみで日付ベースのローテーションがない。

ローテーションは各パイプライン実行後に _rotate_scanned() で実行する。

9.2 データファイル整理

9.3 scanned.json の新フォーマット

{
  "articles": {
    "a1b2c3d4e5f6": {
      "title": "...",
      "url": "...",
      "source_id": "openai_blog",
      "domain_id": "ai",
      "scanned_at": "2026-04-07T07:00:00+09:00",
      "notified": true,
      "evaluation": { ... }
    }
  }
}

現行フォーマットとの違いは domain_id と notified の追加のみ。後方互換性あり（既存エントリは domain_id がなければ "ai" とみなす）。

10. ディレクトリ構成

ops/news_reminder/
├── __init__.py
├── app/
│   ├── __init__.py
│   ├── config.py                 変更  Settings拡張
│   ├── pipeline.py               新規  共通パイプライン
│   ├── llm.py                    新規  Claude CLI統一ラッパー
│   ├── domain.py                 新規  DomainConfig ローダー
│   ├── models.py                 新規  Article, Evaluation, PipelineResult
│   ├── scanner.py                変更  段階的に縮小（最終的に廃止）
│   ├── thread.py                 変更  ヘッダ形式変更
│   ├── dashboard.py              維持  月次処理（変更なし）
│   ├── sources/                  新規
│   │   ├── __init__.py                   アダプターレジストリ
│   │   ├── base.py                       SourceAdapter Protocol
│   │   ├── rss.py                        RssAdapter
│   │   ├── hn.py                         HackerNewsAdapter
│   │   ├── arxiv.py                      ArxivAdapter
│   │   └── google_news.py                GoogleNewsAdapter
│   └── formatters/               新規
│       ├── __init__.py                   フォーマッタレジストリ
│       ├── base.py                       BaseFormatter
│       ├── default.py                    DefaultFormatter（地政学等）
│       ├── ai.py                         AiFormatter（MNML改修余地）
│       └── finance.py                    FinanceFormatter（銘柄影響表）
├── pipeline/                     段階的廃止
│   ├── __init__.py                       Phase 2で廃止
│   ├── __main__.py                       CLIは維持（内部を新パイプラインに差し替え）
│   ├── cli.py                            CLIは維持
│   ├── config.py                         Phase 2で廃止
│   └── steps/                            Phase 2で廃止
│       ├── fetch.py                      → sources/google_news.py に移植
│       ├── summarize.py                  → llm.py + pipeline.py に統合
│       ├── notify.py                     → formatters/finance.py に移植
│       └── sync.py                       維持（Excel同期は独立CLIコマンド）
├── data/
│   ├── sources.json              変更  id追加、category廃止
│   ├── holdings.json             維持
│   ├── scanned.json              変更  domain_id追加、ローテーション
│   ├── domains/                  新規
│   │   ├── ai.json
│   │   ├── finance.json
│   │   └── geopolitics.json
│   └── prompts/                  新規
│       ├── evaluate_ai.txt
│       ├── evaluate_finance.txt
│       ├── evaluate_geopolitics.txt
│       ├── summarize_ai.txt
│       ├── summarize_finance.txt
│       └── summarize_geopolitics.txt
└── tests/                        新規
    ├── __init__.py
    ├── conftest.py                       共通フィクスチャ
    ├── test_sources.py                   ソースアダプターのテスト
    ├── test_pipeline.py                  パイプライン統合テスト
    ├── test_formatters.py                フォーマッタのテスト
    ├── test_llm.py                       LLMラッパーのテスト
    └── test_domain.py                    ドメインローダーのテスト

11. インターフェース定義

11.1 データモデル — app/models.py

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime


@dataclass
class Article:
    """取得した記事"""
    id: str
    title: str
    url: str
    source_name: str
    source_id: str
    published: datetime | None = None
    content: str = ""
    tier: str = "primary"
    metadata: dict = field(default_factory=dict)
    evaluation: Evaluation | None = None


@dataclass
class Evaluation:
    """LLMによる記事評価"""
    relevant: bool = False
    summary_ja: str = ""
    impact: str = ""           # "high" | "medium" | "low"
    action: str = ""           # 推奨アクション
    reason: str = ""           # 評価理由
    extra: dict = field(default_factory=dict)  # ドメイン固有（urgency等）


@dataclass
class PipelineResult:
    """パイプライン実行結果"""
    domain_id: str
    total_fetched: int = 0
    total_new: int = 0
    total_relevant: int = 0
    summary_text: str = ""
    articles: list[Article] = field(default_factory=list)
    error: str | None = None

11.2 ドメインローダー — app/domain.py

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path

DOMAINS_DIR = Path(__file__).resolve().parent.parent / "data" / "domains"
PROMPTS_DIR = Path(__file__).resolve().parent.parent / "data" / "prompts"


@dataclass
class DomainConfig:
    """ドメイン設定"""
    id: str
    name: str
    emoji: str
    thread_prefix: str
    sources: list[str]
    source_config: dict = field(default_factory=dict)
    evaluate_prompt_path: str = ""
    summarize_prompt_path: str = ""
    formatter: str = "default"
    max_articles_per_run: int = 30
    mention_ceo: bool = True

    def load_evaluate_prompt(self) -> str:
        """評価プロンプトテンプレートを読み込む"""
        ...

    def load_summarize_prompt(self) -> str:
        """要約プロンプトテンプレートを読み込む"""
        ...


def load_domain(domain_id: str) -> DomainConfig:
    """指定ドメインの設定を読み込む"""
    ...


def load_all_domain_ids() -> list[str]:
    """data/domains/ 内の全ドメインIDを返す"""
    ...

11.3 ソースアダプター — app/sources/base.py

from __future__ import annotations

from typing import Protocol

from news_reminder.app.domain import DomainConfig
from news_reminder.app.models import Article


class SourceAdapter(Protocol):
    """ソースアダプターのプロトコル"""

    async def fetch(
        self,
        source_config: dict,
        domain_config: DomainConfig,
    ) -> list[Article]:
        ...


# アダプターレジストリ
_ADAPTERS: dict[str, type] = {}

def register(source_type: str):
    """デコレータ: ソースタイプにアダプターを登録"""
    def decorator(cls):
        _ADAPTERS[source_type] = cls
        return cls
    return decorator

def get_adapter(source_type: str) -> SourceAdapter:
    """ソースタイプに対応するアダプターを返す"""
    cls = _ADAPTERS.get(source_type)
    if cls is None:
        raise ValueError(f"未知のソースタイプ: {source_type}")
    return cls()

11.4 フォーマッタ — app/formatters/base.py

from __future__ import annotations

from typing import Protocol

from news_reminder.app.domain import DomainConfig
from news_reminder.app.models import Article, PipelineResult


class Formatter(Protocol):
    """Slack投稿フォーマッタのプロトコル"""

    def build_blocks(
        self,
        result: PipelineResult,
        domain: DomainConfig,
    ) -> tuple[list[dict], str]:
        """Block Kitブロックとフォールバックテキストを返す"""
        ...

11.5 Settings拡張 — app/config.py

class Settings(BaseSettings):
    """環境変数から読み込む設定"""

    # Slack
    slack_bot_token: str = ""
    slack_channel: str = "news-reminder"

    # Claude CLI
    claude_cli_path: str = "claude"
    default_model: str = "haiku"
    llm_timeout: int = 120
    evaluate_batch_size: int = 20

    # データ管理
    scanned_max_articles: int = 1000
    scanned_max_days: int = 90

    # CEO
    ceo_user_id: str = "U0AHXTRDQMA"

    model_config = {"env_file": ".env", "env_file_encoding": "utf-8", "extra": "ignore"}

12. 移行計画

Phase 1: 基盤モジュール作成

既存コードに影響を与えず、新モジュールを追加する。

• app/models.py — データモデル定義
• app/domain.py — ドメインローダー
• app/llm.py — Claude CLI統一ラッパー
• app/sources/ — ソースアダプター（既存コードから関数を移植）
• data/domains/{ドメイン名}.json — ドメイン設定ファイル（ai.json, finance.json, geopolitics.json）
• data/prompts/*.txt — プロンプトテンプレート（現行scanner.pyのプロンプトを抽出）
• tests/ — 新モジュールのテスト

Phase 2: パイプライン統合

新パイプラインに切り替え、旧コードを廃止する。

• app/pipeline.py — 共通パイプライン実装
• app/formatters/ — Slack投稿フォーマッタ
• bot/scheduler.py — _run_all_domains() に切り替え
• pipeline/__main__.py / cli.py — 内部を新パイプラインに差し替え
• scanner.py の取得・評価ロジックを削除（build_summary_blocks 等のscheduler依存関数は残す）
• pipeline/steps/ — 廃止（sync.pyのみCLIコマンドとして残す）
• scanned.json ローテーション実装
• notified.json → scanned.json への統合マイグレーション

Phase 3: 旧コード削除・最終整理

• scanner.py 完全廃止（build_summary_blocks 等はformatterに移行済み）
• pipeline/config.py 廃止（PipelineConfig / Holding はmodels.py / domain.py に統合済み）
• fetched.json / notified.json 削除
• sources.json の旧 category フィールド削除
• scheduler.pyの旧ジョブ（news_scan / news_pipeline_*）を削除
• launchd plist（com.news-reminder.pipeline）を廃止

13. テスト戦略

13.1 テスト対象と方針

13.2 テストで検証しないもの

• 実際のClaude CLI呼び出し（モックで代替）
• 実際のSlack API呼び出し（モックで代替）
• 外部RSSフィードの可用性

14. セキュリティ・パフォーマンス

14.1 セキュリティ

14.2 パフォーマンス

14.3 キャッシュ戦略

前提条件・制約

• X(Twitter) API対応は設計上のソースタイプ定義のみ。実装はresearcherの調査結果（Grok API / X API v2）が揃ってから
• Claude CLIはMax Plan（API課金なし）。Haiku使用でコスト最適化
• dashboard.py（BlackRock月次解釈）は今回のスコープ外。現行のまま維持
• pipeline/steps/sync.py（Excel銘柄同期）は独立CLIコマンドとして維持
• IF層への候補通知（AIドメインのみ）は現行ロジックを scheduler.py 側に維持