CAN-SLIM の C｜四半期 EPS 急成長を Python で自動判定する手順とサンプルコード入門

前回の記事#21では、発展編シリーズ（#21〜#30）の入口として、CAN-SLIM 7要素の概要と応用編との戦略的差別化、発展編ロードマップを整理しました。本記事から CAN-SLIM の各要素を Python で個別に実装していくサンプルコード入門が始まります。最初は C（Current Quarterly EPS、四半期 EPS の前年同期比 急成長）です。

CAN-SLIM の C 要素は 「四半期 EPS が前年同期比で +25% 以上の急成長を示す銘柄」を抽出する基準です。「直近の業績加速度」を捉える指標で、CAN-SLIM 7要素の中で最も自動判定しやすく、本記事を最初に置いたのも実装の取りかかりやすさからです。絶対値より変化率、突発より連続性という評価設計原則が本記事の核心で、本業のプロトタイプ性能評価でも繰り返し使ってきた発想です。

多くの読者がぶつかる壁は次の3つです:

• 四半期 EPS データの取得: 日本株は J-Quants の /fins/statements、米国株は FMP の /income-statement/{symbol}。両者で API 仕様・カラム名・更新頻度が異なる
• 前年同期比（YoY）の計算: 「同じ第3四半期の昨年と比較」する必要があるが、決算月のズレ（3月期 vs 12月期 vs 9月期）で実装が複雑化。さらに赤字→黒字転換時の分母負値問題
• 応用編パイプラインへの統合: 応用編 #15 の screen_parallel はインカム重視の高配当株判定。成長株版の判定を追加するときの干渉を避ける設計

筆者は製造業の研究開発現場で、新技術プロトタイプの直近性能評価を多く経験してきました。「3ヶ月前比 +25% 改善」を判定基準にしてプロトタイプの量産化判定を行う作法は、CAN-SLIM C の「直近の成長加速度を見る」発想と構造的に対応します。本業の経験を投資データに転用する典型例として、本記事のエピソードで詳しく整理します。

本記事では 4 個のサンプルコードで、四半期 EPS 取得から CAN-SLIM C 要素の自動判定、応用編 #15 のスクリーニング Runner への統合まで実装します。次回 #23 では A（年間 EPS 成長率の3-5年トレンド可視化）に進みます。

結論：CAN-SLIM の C 要素は「四半期 EPS の前年同期比 +25% 以上 + 連続2-3期の継続性」を機械的に自動判定する。データ取得は J-Quants（日本株）と FMP（米国株）の両 API で対応、YoY 計算は pandas の groupby + shift で1年前同期との比較、判定は応用編 #15 の判定器フレームワークを再利用して growth_can_slim 戦略として組み込む。本記事の実装が完了すると、応用編パイプラインの上に CAN-SLIM C 要素が動く状態になり、発展編 #23 以降の他要素を順次追加していく土台が整う。「絶対値より変化率、突発より連続性」が本記事を貫く評価設計原則。

CAN-SLIM C 要素の役割と「+25%」閾値の意味（急成長判定の手順）

CAN-SLIM の C は Current Quarterly Earnings Per Share、四半期 EPS の急成長を見ます。オニール原典では次の3点が中核:

• 当期 Q の EPS が前年同期比 +25% 以上の急成長
• サプライズ要因（一過性の利益）ではなく、本業の成長を反映（絶対値より変化率を見る）
• 過去2-3四半期の連続成長傾向との整合（突発より連続性を見る、直近1Q だけ突発的に伸びるのではなく）

スニペット1：四半期 EPS データの取得サンプルコード入門（J-Quants + FMP の使い分け）

日本株と米国株では API が異なるため、抽象化レイヤーを設けて両方を統一フォーマットで取得します。J-Quants は応用編 #13 で扱った認証フローを再利用、FMP は apikey パラメータでシンプルに取得できます。J-Quants の TypeOfDocument フィルタは TypeOfCurrentPeriod を使った正確な絞り込みに修正しました（v1 の文字列部分一致だと公式仕様に不整合）。

# fetch_quarterly_eps.py — 日米両市場の四半期 EPS を統一フォーマットで取得（v2: J-Quants フィルタ修正）
# 動作環境: Python 3.11+ / pandas 2.x / requests 2.31+
import os
import requests
import pandas as pd
from typing import Literal

JQUANTS_BASE = "https://api.jquants.com/v1"
FMP_BASE = "https://financialmodelingprep.com/api/v3"

def fetch_quarterly_eps_jp(id_token: str, code: str) -> pd.DataFrame:
    """日本株の四半期 EPS を取得（J-Quants /fins/statements）

    返却カラム:
    ticker / period_end / fiscal_quarter (1Q/2Q/3Q) / eps / market='JP'

    v2 修正: TypeOfDocument 部分一致 → TypeOfCurrentPeriod の isin で正確に絞り込む
    """
    headers = {"Authorization": f"Bearer {id_token}"}
    params = {"code": code}
    resp = requests.get(f"{JQUANTS_BASE}/fins/statements", headers=headers, params=params, timeout=60)
    resp.raise_for_status()
    rows = resp.json().get("statements", [])
    df = pd.DataFrame(rows)
    if df.empty:
        return pd.DataFrame(columns=["ticker", "period_end", "fiscal_quarter", "eps", "market"])
    # TypeOfCurrentPeriod が "1Q" / "2Q" / "3Q" のどれかにマッチする四半期決算のみ抽出
    quarterly = df[df["TypeOfCurrentPeriod"].isin(["1Q", "2Q", "3Q"])]
    return pd.DataFrame({
        "ticker": quarterly["LocalCode"].str[:4],
        "period_end": pd.to_datetime(quarterly["CurrentPeriodEndDate"]),
        "fiscal_quarter": quarterly["TypeOfCurrentPeriod"],  # 1Q / 2Q / 3Q
        "eps": pd.to_numeric(quarterly["EarningsPerShare"], errors="coerce"),
        "market": "JP",
    })

def fetch_quarterly_eps_us(symbol: str, fmp_apikey: str, periods: int = 12) -> pd.DataFrame:
    """米国株の四半期 EPS を取得（FMP /income-statement/{symbol} 四半期版）

    返却カラム:
    ticker / period_end / fiscal_quarter (Q1/Q2/Q3/Q4) / eps / market='US'

    注意: FMP の eps は GAAP ベース。Non-GAAP は別エンドポイント
    """
    url = f"{FMP_BASE}/income-statement/{symbol}"
    params = {"period": "quarter", "limit": periods, "apikey": fmp_apikey}
    resp = requests.get(url, params=params, timeout=60)
    resp.raise_for_status()
    rows = resp.json()
    df = pd.DataFrame(rows)
    if df.empty:
        return pd.DataFrame(columns=["ticker", "period_end", "fiscal_quarter", "eps", "market"])
    return pd.DataFrame({
        "ticker": symbol,
        "period_end": pd.to_datetime(df["date"]),
        "fiscal_quarter": df["period"],  # FMP は "Q1" / "Q2" / "Q3" / "Q4"
        "eps": pd.to_numeric(df["eps"], errors="coerce"),
        "market": "US",
    })

def fetch_quarterly_eps(market: Literal["JP", "US"], identifier: str, **kwargs) -> pd.DataFrame:
    """統一インターフェイス: market を指定して日本株 or 米国株の四半期 EPS を取得"""
    if market == "JP":
        id_token = kwargs.get("id_token") or os.environ["JQUANTS_ID_TOKEN"]
        return fetch_quarterly_eps_jp(id_token=id_token, code=identifier)
    elif market == "US":
        apikey = kwargs.get("fmp_apikey") or os.environ["FMP_APIKEY"]
        return fetch_quarterly_eps_us(symbol=identifier, fmp_apikey=apikey)
    else:
        raise ValueError(f"Unsupported market: {market}")

# 使用例（架空銘柄）
# df_jp = fetch_quarterly_eps("JP", "XXXX")  # 日本株の4桁コード
# df_us = fetch_quarterly_eps("US", "AAAA")  # 米国株のティッカー

スニペット2：前年同期比（YoY）成長率の計算と DuckDB 保存

四半期 EPS データを取得したら、同じ四半期の1年前データと比較して YoY 成長率を計算します。pandas の groupby + shift(1) を使い、ticker・fiscal_quarter ごとに1本前（同じ四半期の1年前）の値を取り出して比較するパターンです。「絶対値より変化率を見る」の絶対値分母採用、「突発より連続性を見る」の連続増益四半期数も同時計算します。

# compute_eps_yoy.py — 四半期 EPS の YoY 成長率を計算 + DuckDB 保存（v2: 連続性判定追加 + DDL 完成）
# 動作環境: Python 3.11+ / pandas 2.x / numpy / duckdb 1.0+
import duckdb
import pandas as pd
import numpy as np
from pathlib import Path

DB_PATH = Path("data/stocks.duckdb")

DDL_QUARTERLY_EPS_WITH_YOY = """
CREATE TABLE IF NOT EXISTS quarterly_eps_with_yoy (
    ticker_normalized           VARCHAR NOT NULL,
    period_end                  DATE NOT NULL,
    fiscal_quarter              VARCHAR NOT NULL,
    market                      VARCHAR NOT NULL,
    eps                         DOUBLE,
    eps_prev_year               DOUBLE,
    yoy_growth_pct              DOUBLE,         -- 絶対値分母 YoY (%)
    yoy_growth_pct_capped       DOUBLE,         -- 前年正値時のみ
    consecutive_up_quarters     INTEGER,        -- 連続増益四半期数（突発vs連続性判定）
    computed_at                 TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (ticker_normalized, period_end)
);
"""

def compute_quarterly_eps_yoy(df: pd.DataFrame) -> pd.DataFrame:
    """四半期 EPS の前年同期比（YoY）成長率と連続増益四半期数を計算

    入力 df: ticker / period_end / fiscal_quarter / eps / market
    返却に追加するカラム:
    - eps_prev_year: 1年前同期の EPS（同じ fiscal_quarter ラベル内で1本前）
    - yoy_growth_pct: 前年同期比成長率 (%、絶対値分母)
    - yoy_growth_pct_capped: 異常値（赤字→黒字転換等）を抑制した値
    - consecutive_up_quarters: 連続して YoY > 0 の四半期数（連続性判定）
    """
    df = df.sort_values(["ticker", "fiscal_quarter", "period_end"]).copy()
    # 同じ ticker × 同じ fiscal_quarter で1本前（= 1年前同期）の EPS を取得
    df["eps_prev_year"] = df.groupby(["ticker", "fiscal_quarter"])["eps"].shift(1)

    # YoY 計算: ((当期 - 1年前) / |1年前|) × 100。絶対値分母で「絶対値より変化率」原則を担保
    valid = df["eps_prev_year"].notna() & (df["eps_prev_year"] != 0)
    df["yoy_growth_pct"] = np.where(
        valid,
        ((df["eps"] - df["eps_prev_year"]) / df["eps_prev_year"].abs()) * 100,
        np.nan,
    )
    df["yoy_growth_pct_capped"] = np.where(
        df["eps_prev_year"] > 0, df["yoy_growth_pct"], np.nan
    )

    # 連続増益四半期数: ticker ごとに period_end 順に並べて、yoy > 0 の連続を数える
    df = df.sort_values(["ticker", "period_end"])
    df["is_up"] = (df["yoy_growth_pct_capped"] > 0).astype(int)
    df["streak_group"] = (df["is_up"] != df.groupby("ticker")["is_up"].shift(1)).cumsum()
    df["consecutive_up_quarters"] = df.groupby(["ticker", "streak_group"]).cumcount() + 1
    df.loc[df["is_up"] == 0, "consecutive_up_quarters"] = 0
    df = df.drop(columns=["is_up", "streak_group"])
    return df

def upsert_quarterly_eps_with_yoy(df: pd.DataFrame) -> None:
    """compute_quarterly_eps_yoy の出力を idempotent に保存"""
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    target = df.rename(columns={"ticker": "ticker_normalized"})[
        ["ticker_normalized", "period_end", "fiscal_quarter", "market",
         "eps", "eps_prev_year", "yoy_growth_pct", "yoy_growth_pct_capped",
         "consecutive_up_quarters"]
    ]
    with duckdb.connect(str(DB_PATH)) as conn:
        conn.execute(DDL_QUARTERLY_EPS_WITH_YOY)
        conn.register("eps_df", target)
        conn.execute("BEGIN")
        try:
            conn.execute("""
                DELETE FROM quarterly_eps_with_yoy
                WHERE (ticker_normalized, period_end) IN
                      (SELECT ticker_normalized, period_end FROM eps_df)
            """)
            conn.execute("""
                INSERT INTO quarterly_eps_with_yoy
                    (ticker_normalized, period_end, fiscal_quarter, market,
                     eps, eps_prev_year, yoy_growth_pct, yoy_growth_pct_capped,
                     consecutive_up_quarters)
                SELECT * FROM eps_df
            """)
            conn.execute("COMMIT")
        except Exception:
            conn.execute("ROLLBACK")
            raise
        finally:
            conn.unregister("eps_df")

# 使用例
# from fetch_quarterly_eps import fetch_quarterly_eps
# df = fetch_quarterly_eps("JP", "XXXX")
# df = compute_quarterly_eps_yoy(df)
# upsert_quarterly_eps_with_yoy(df)

スニペット3：CAN-SLIM C 要素の判定器（連続性判定込みの自動判定サンプルコード）

YoY 成長率と連続増益四半期数が計算できたら、「+25% 以上 + 連続2期以上で PASS」「+10% 以上で CAUTION」「それ未満で FAIL」の3段階判定を実装します。応用編 #15 の spec_sheet_judge_v2 と構造的に対応するパターンで、C 要素単独の判定器を作ります。「突発より連続性を見る」原則を min_consecutive 引数で実装します。

# can_slim_c_judge.py — CAN-SLIM C 要素の判定器（v2: 連続性判定追加）
# 動作環境: Python 3.11+ / pandas 2.x / numpy
from typing import Literal, TypedDict, NotRequired
import numpy as np

Verdict = Literal["PASS", "CAUTION", "FAIL"]

class CElementMetrics(TypedDict, total=False):
    yoy_growth_pct: NotRequired[float | None]
    consecutive_up_quarters: NotRequired[int | None]

def judge_c_element(metrics: CElementMetrics,
                     threshold_pass: float = 25.0,
                     threshold_caution: float = 10.0,
                     min_consecutive: int = 2) -> Verdict:
    """CAN-SLIM C 要素を3段階で自動判定（v2: 連続性判定追加で「突発より連続性」原則を担保）

    threshold_pass=25.0 (原典準拠), threshold_caution=10.0
    min_consecutive=2 (連続2期以上の増益で「本物の急成長」と判定、原典準拠)

    赤字→黒字転換などで yoy_growth_pct が None の場合は CAUTION
    """
    yoy = metrics.get("yoy_growth_pct")
    if yoy is None:
        return "CAUTION"
    if yoy < threshold_caution:
        return "FAIL"
    if yoy < threshold_pass:
        return "CAUTION"
    # +25% 以上を満たした上で、連続性チェック（突発より連続性）
    consec = metrics.get("consecutive_up_quarters") or 0
    if consec < min_consecutive:
        return "CAUTION"  # 急成長だが連続性なし、突発の可能性
    return "PASS"

def screen_c_element(eps_with_yoy_df: "pd.DataFrame",
                      threshold_pass: float = 25.0,
                      threshold_caution: float = 10.0,
                      min_consecutive: int = 2) -> "pd.DataFrame":
    """DataFrame に対して C 要素判定をベクトル化適用（v2: 引数化 + hasattr で AttributeError 回避）"""
    import pandas as pd
    df = eps_with_yoy_df.copy()
    # capped が無い場合は yoy_growth_pct にフォールバック（hasattr で安全に判定）
    if "yoy_growth_pct_capped" in df.columns:
        yoy = df["yoy_growth_pct_capped"]
    elif "yoy_growth_pct" in df.columns:
        yoy = df["yoy_growth_pct"]
    else:
        df["c_verdict"] = "CAUTION"
        return df

    consec = df["consecutive_up_quarters"] if "consecutive_up_quarters" in df.columns else pd.Series(0, index=df.index)

    # ベクトル化判定
    df["c_verdict"] = np.where(
        yoy.isna(), "CAUTION",
        np.where(yoy < threshold_caution, "FAIL",
                 np.where(yoy < threshold_pass, "CAUTION",
                          np.where(consec < min_consecutive, "CAUTION", "PASS"))),
    )
    return df

# 使用例
# from compute_eps_yoy import compute_quarterly_eps_yoy
# from fetch_quarterly_eps import fetch_quarterly_eps
#
# df = fetch_quarterly_eps("JP", "XXXX")
# df = compute_quarterly_eps_yoy(df)
# df = screen_c_element(df, threshold_pass=25.0, min_consecutive=2)
#
# latest = df.sort_values("period_end").groupby("ticker").tail(1)
# print(latest[["ticker", "yoy_growth_pct_capped", "consecutive_up_quarters", "c_verdict"]])

スニペット4：応用編 #15 の screen_parallel に CAN-SLIM C を統合する手順

応用編 #15 の screen_parallel.py は高配当株判定がメインでした。本記事で C 要素判定を追加し、1回の Runner 起動で「高配当 PASS / 成長 C 要素 PASS」両方を出力できるように拡張します。応用編 #19 で導入した portfolio_scores や、発展編 #21 の strategy_criteria テーブルとも組み合わさる設計です。

v2 では SQL を ROW_NUMBER OVER で「直近期の YoY」を正しく取得する形に修正しています（v1 の MAX は「過去最高 YoY」を返してしまう誤実装でした）。

# screen_parallel_v2.py — 高配当 + CAN-SLIM C を1パイプラインで判定（v2: 直近期 YoY を ROW_NUMBER で取得）
# 動作環境: Python 3.11+ / duckdb 1.0+ / pandas 2.x
import duckdb
import pandas as pd
from multiprocessing import Pool, cpu_count
from datetime import date
from pathlib import Path
from spec_sheet_judge_v2 import judge_v2  # 応用編 #15 から継承
from can_slim_c_judge import judge_c_element  # 本記事で追加

DB_PATH = Path("data/stocks.duckdb")
OUT_DIR = Path("data/screening_results")

_RULES_BY_PROFILE: dict = {}
_C_THRESHOLD: float = 25.0
_C_MIN_CONSECUTIVE: int = 2

def _init_worker(rules_by_profile: dict, c_threshold: float, c_min_consecutive: int) -> None:
    """initializer パターン（応用編 #15 で確立、spawn 環境対応）"""
    global _RULES_BY_PROFILE, _C_THRESHOLD, _C_MIN_CONSECUTIVE
    _RULES_BY_PROFILE = rules_by_profile
    _C_THRESHOLD = c_threshold
    _C_MIN_CONSECUTIVE = c_min_consecutive

def judge_one_ticker(metrics_with_eps: dict) -> dict:
    """1銘柄に対して 高配当判定 + CAN-SLIM C 判定 の両方を実施"""
    profile = metrics_with_eps.get("industry_profile") or "manufacturing"
    rules = _RULES_BY_PROFILE.get(profile, _RULES_BY_PROFILE.get("manufacturing", {}))
    high_dividend_result = judge_v2(metrics_with_eps, rules)

    c_result = judge_c_element(
        {"yoy_growth_pct": metrics_with_eps.get("yoy_growth_pct"),
         "consecutive_up_quarters": metrics_with_eps.get("consecutive_up_quarters")},
        threshold_pass=_C_THRESHOLD,
        min_consecutive=_C_MIN_CONSECUTIVE,
    )

    return {
        "ticker_normalized": metrics_with_eps["ticker_normalized"],
        "name_jp": metrics_with_eps.get("name_jp"),
        "industry_profile": profile,
        "high_dividend_overall": high_dividend_result["overall"],
        "can_slim_c_verdict": c_result,
        "yoy_growth_pct": metrics_with_eps.get("yoy_growth_pct"),
        "consecutive_up_quarters": metrics_with_eps.get("consecutive_up_quarters"),
    }

def run_screening_v2(c_threshold: float = 25.0, c_min_consecutive: int = 2) -> Path:
    """高配当 + CAN-SLIM C を並列で自動判定して CSV/Parquet 出力"""
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    today = date.today().isoformat()
    out_path = OUT_DIR / f"{today}_v2.parquet"

    with duckdb.connect(str(DB_PATH)) as conn:
        rules: dict = {}
        rows = conn.execute("""
            SELECT industry_profile, standard_indicator, direction, threshold_pass, threshold_caution
            FROM industry_indicator_map
        """).fetchall()
        for prof, ind, dirn, p, c in rows:
            rules.setdefault(prof, {})[ind] = {
                "direction": dirn, "threshold_pass": p, "threshold_caution": c,
                "range_high_pass": None, "range_high_caution": None,
            }

        # v2 修正: ROW_NUMBER OVER で「銘柄ごとの直近期 YoY」を取得（MAX だと過去最高になる誤実装の修正）
        df = conn.execute("""
            SELECT s.*, e.yoy_growth_pct, e.consecutive_up_quarters
            FROM v_screening_input s
            LEFT JOIN (
                SELECT ticker_normalized, yoy_growth_pct_capped AS yoy_growth_pct,
                       consecutive_up_quarters
                FROM (
                    SELECT *,
                        ROW_NUMBER() OVER (PARTITION BY ticker_normalized
                                            ORDER BY period_end DESC) AS rn
                    FROM quarterly_eps_with_yoy
                ) ranked
                WHERE rn = 1
            ) e ON s.ticker_normalized = e.ticker_normalized
        """).fetchdf()
    metrics_list = df.to_dict(orient="records")

    with Pool(processes=cpu_count(), initializer=_init_worker,
               initargs=(rules, c_threshold, c_min_consecutive)) as pool:
        results = pool.map(judge_one_ticker, metrics_list, chunksize=50)

    result_df = pd.DataFrame(results)
    result_df.to_parquet(out_path, index=False)
    print(f"saved: {out_path}, rows={len(result_df)}")
    print(f"  high_dividend PASS: {(result_df['high_dividend_overall']=='PASS').sum()}")
    print(f"  CAN-SLIM C PASS:   {(result_df['can_slim_c_verdict']=='PASS').sum()}")
    print(f"  両方 PASS:         {((result_df['high_dividend_overall']=='PASS') & (result_df['can_slim_c_verdict']=='PASS')).sum()}")
    return out_path

if __name__ == "__main__":
    run_screening_v2(c_threshold=25.0, c_min_consecutive=2)

設計判断の記録：CAN-SLIM C 実装の3判断 + 応用編 #13〜#21 + 発展編 #22 の判断俯瞰表

判断0：CAN-SLIM C 実装で記録すべき3判断

本記事は CAN-SLIM の最初の要素実装として、シリーズ全体の方針に関わる設計判断を3つ記録します。判断1（YoY 計算ロジック）は数値の意味、判断2（赤字→黒字転換の扱い）は欠損戦略、判断3（screen_parallel 統合方法）は応用編との連携を左右します。

判断1：YoY 計算で「絶対値ベース分母」を使うか「単純前期比」「対数差」にするか

• 採用理由: 絶対値分母 (当期 - 前年) / |前年| を採用。代替案との比較:
  • 単純前期比 (当期 - 前年) / 前年: 数学的に厳密だが、前年が負値→当期が更に大きな負値（赤字拡大）のとき方向性が反転する罠
  • 対数差 log(当期) - log(前年): 統計学的に綺麗だが、負値が扱えず赤字銘柄でゼロ除算と同等の問題
  • 絶対値分母（採用）: 「赤字幅が縮小した（改善）」を正の値で表現できる、判定器側の if 分岐がシンプル
• 採用したことで失うもの: 数学的に厳密な「前年比」（負値分母）の表現
• トリガー条件: 株式アナリストレポートで前年比を機械的に流用したい場合
• 残るメリット: 機械判定で「赤字幅縮小（改善）」を正の値で扱える、別カラム yoy_growth_pct（絶対値分母）と yoy_growth_pct_capped（前年正値時のみ）の2つを用途で使い分け
• 対処: 厳密な単純前期比が必要な場合は別カラム（追加）で対応
• 本業 DX 同型対応: 製造業のプロセス改善でも「前期比改善率」は分母に絶対値を使うのが標準作法

判断2：赤字→黒字転換銘柄を CAUTION とするか PASS とするか

• 採用理由: 機械判定では CAUTION（要再確認）扱い。赤字→黒字転換は「ターンアラウンド」と呼ばれる別カテゴリで、CAN-SLIM C の「成長加速度」とは別ロジックが必要
• 採用したことで失うもの: ターンアラウンド銘柄の機械抽出機会
• トリガー条件: 投資スタイルが「ターンアラウンド株」中心の場合
• 残るメリット: CAN-SLIM C の本来の意図（連続成長中の銘柄）に絞った精度確保
• 対処: 別途「turnaround_screening.py」を発展編 #29 のハイブリッド戦略あたりで検討
• 本業 DX 同型対応: 製造業の品質評価でも「不良→良品転換ライン」と「常時良品ライン」は別評価枠組みで管理するのが定石

判断3：screen_parallel に統合するか別 Runner にするか

• 採用理由: 応用編 #15 の screen_parallel に CAN-SLIM C を組み込むことで、1回の Runner 起動で高配当 + 成長の両判定が出る
• 採用したことで失うもの: 高配当判定だけを高頻度で実行する場合の処理時間（CAN-SLIM C 判定のオーバーヘッド）
• トリガー条件: 高配当判定だけを毎日実行する運用
• 残るメリット: ハイブリッド戦略（発展編 #29）に必要な統合判定が早期に動く状態に
• 対処: 引数で「両判定 / 高配当のみ / CAN-SLIM のみ」を切り替えられる CLI フラグを追加
• 本業 DX 同型対応: 製造業の検査ラインでも「全検査 vs 簡易検査」を切替可能にする設計が一般的

応用編 #13〜#21 + 発展編 #22 の主要設計判断 俯瞰表（26件）

応用編 #17 から指摘されてきた「応用編全体の判断俯瞰表」を、発展編 #21 で初実装し、本記事 #22 で2件追加します。応用編 + 発展編で記録した主要判断を一覧化:

記事	主要判断	採用
#13	データソース選定	J-Quants + EDINET（公式 API 優先）
#14	DB / アーキテクチャ	DuckDB + ELT パターン + マスタ駆動
#14	業種別補正設計	industry_indicator_map テーブル駆動 + direction カラム
#15	並列化技術	multiprocessing（CPU bound、initializer パターン）
#15	通知メディア	LINE Messaging API
#15	スケジューラ	cron → GitHub Actions
#16	連続増配判定	連続非減少（同額据え置きOK）
#16	EPS 安定性指標	変動係数 CV + 線形回帰
#16	新規上場銘柄の扱い	5期未満は CAUTION 強制
#17	異常値検知の組合せ	IQR + Z + SPC + WER の3-4層フィルタ
#17	市場全体ショック対応	絶対値 + #20 で TOPIX 相対値併用
#18	業種粒度	10業種 → #19 で15 → #20 で25業種
#18	業種別主指標数	1業種1主指標
#18	マスタ更新サイクル	四半期レビュー
#18	テーブル PK 設計	industry_indicator_map: 複合PK / industry_cv_threshold: 単列PK
#19	集中度指標	HHI（投資視点閾値 2000/3000）
#19	相関分析期間	過去24ヶ月（月次） + 直近6ヶ月併用
#19	FMEA RPN 重み	等倍積（S × O × D）
#19	ポートフォリオテーブル	個別銘柄評価とは別テーブル
#20	master_runner 方式	NotImplementedError による委譲
#20	業種カバレッジ	25業種で東証時価総額の概ね9割
#20	市場ベースライン	TOPIX 相対変動率（Premium プラン要、別ソース対応も）
#21	成長株フレームワーク	CAN-SLIM（7要素中6要素が定量化可能）
#21	パイプライン再利用	応用編の主要要素を流用、判定層のみ拡張
#21	市場カバレッジ	日米両市場対応（FMP + SEC EDGAR 追加）
#22	YoY 計算ロジック	絶対値分母（赤字→黒字転換時の方向性反転を回避）
#22	連続性判定	min_consecutive=2 で「突発より連続性」原則を担保

本業の話：プロトタイプの3ヶ月性能評価で +25% 改善 + 連続2-3期を判定基準にした経験

筆者が製造業の研究開発部門で新技術プロトタイプの量産化判定に関わっていたとき、ベテランから次のように指導されました:

• 「プロトタイプの『絶対性能』だけ見るな。直近3ヶ月の性能改善率を見ろ。改善率が +25% 以上なら『成長中』、+10〜25%なら『安定期』、+10% 未満なら『成熟』と判定する」
• 「サプライズ要因（一時的なベンチマーク条件改善）と区別しろ。連続2-3期で +25% 以上なら本物の成長加速度、1期だけ突発的なら一過性の可能性」
• 「赤字→黒字転換の判定は別ロジック。改善率を機械的に計算すると意味のない数値になる。ターンアラウンド評価は別フレームワークで」

具体的な業務インパクトと3年スパンのマイルストーン:

• 1年目 Q1（指摘）: 初稿の判定基準では「絶対性能が一定値を超えた」プロトタイプを量産化候補にしていた。ベテラン指摘で「+25% 改善率 + 連続2-3期」基準に切替提案、社内予算は維持
• 1年目 Q2-Q4（改修）: 評価基準テーブルを更新し、過去6ヶ月分のプロトタイプ評価を再実行。量産化判定の精度が約3割向上（量産後の市場性能の予測精度向上で測定）。連続性判定で「1期突発的に伸びた偶然のプロトタイプ」を量産候補から除外できるようになる
• 2年目（量産化）: 改修した評価基準で量産化判定を実施し、量産品の初期不良率が予測通り0.3%に収まる。「絶対値より変化率、突発より連続性」原則が部内で標準化
• 3年目（市場投入結果）: 市場投入後のリコール件数が前年比約半減。連続性判定で除外した「1期突発の偶然プロトタイプ」がもし量産化されていれば、市場投入後にリコールに発展するパターンが多かったと事後分析で判明

本記事の CAN-SLIM C 判定器は、この本業の経験と構造的に対応します:

• +25% を PASS、+10-25% を CAUTION、それ未満を FAIL とする3段階判定
• 赤字→黒字転換は CAUTION 扱いで、別ロジック（ターンアラウンド評価）を推奨
• 連続性判定（min_consecutive=2）を実装し、「突発より連続性」原則をコードレベルで担保

逆方向の転移：投資の YoY 計算が本業のプロトタイプ評価を強化

本記事の YoY 計算（特に絶対値分母を使うパターン）と連続性判定を整理した結果、本業のプロトタイプ評価でも同等の改善が期待されます。本業ではこれまで「前期比は単純な (当期-前期)/前期 で計算」していましたが、絶対値分母を使うことで一時的赤字期の混乱を回避できる可能性があります。連続性判定の実装パターン（pandas cumsum + cumcount）も本業の品質管理に逆輸入できる見込みです。応用編で確立した双方向の知識循環が発展編でも続いています。

まとめ：CAN-SLIM C は YoY +25% + 連続2期で機械判定、応用編 #15 → 発展編 #22 の構造的進化

• CAN-SLIM C 要素は「四半期 EPS の前年同期比 +25% 以上 + 連続2期以上」で機械判定。J-Quants（日本株、TypeOfCurrentPeriod でフィルタ）と FMP（米国株、GAAP EPS）の両 API で取得し、pandas の groupby + shift で1年前同期との比較で YoY を算出。赤字→黒字転換は CAUTION 扱いで別ロジック（ターンアラウンド）に送る。「絶対値より変化率、突発より連続性」が本記事を貫く評価設計原則
• 応用編 #15 の screen_parallel に統合することで、1回の Runner 起動で高配当 + 成長 C 要素の両判定が出る。multiprocessing initializer パターン・DuckDB トランザクション境界・strategy_criteria テーブル駆動を維持し、発展編 #23 以降の他要素も同じパターンで追加可能
• 応用編 #15 → 発展編 #22 の構造的進化5点: (1) 1判定 → 2判定（高配当+成長）、(2) 単一閾値 → 連続性判定追加、(3) 業種補正のみ → 戦略軸（高配当 vs 成長）の切替、(4) 1元的 PASS → 複合判定（PASS×PASS）、(5) ハイブリッド戦略の基盤整備

今日からできる3つのアクション

1. 米国株口座の開設と FMP Starter プランの登録（まだなら）。マネックス証券・SBI証券・楽天証券のいずれかで米国株口座を開設し、FMP のサンプルコード入門を試すための準備を整える。年間コスト約3.4万円（FMP $20×12）の支出を確認
2. 本記事のスニペット1〜2を順に実行し、自分の興味のある日本株1銘柄と米国株1銘柄の四半期 EPS と YoY 成長率を計算してみる。直近4期分のデータを取得して、CAN-SLIM C 基準（+25% + 連続2期）に照らした判定が直感と合うか確認
3. スニペット3-4 で応用編 #15 の screen_parallel に CAN-SLIM C を統合し、両判定 PASS 銘柄数 / 高配当のみ PASS 銘柄数 / CAN-SLIM のみ PASS 銘柄数の3つを比較。ハイブリッド戦略（発展編 #29）の候補を洗い出す

次回予告：CAN-SLIM の A — 年間 EPS 成長率の3-5年トレンド可視化（#23→#24→#29 の3段ロケット）

次回（記事#23）では、CAN-SLIM の A（Annual EPS Growth、年間 EPS 成長率の3-5年トレンド）を扱います。本記事の C が「直近1四半期の成長加速度」だったのに対し、A は「過去3-5年の長期成長持続性」を見る指標です。両者を組み合わせる3段ロケットで、発展編シリーズの判定が完成形に近づきます:

• #23 A 要素実装: 年間 EPS データの取得・3-5年成長率計算・matplotlib 可視化
• #24 C×A 複合判定: 短期急成長（C）+ 長期安定成長（A）の組合せで本物の成長株を抽出
• #29 ハイブリッド戦略: 高配当 × 成長 × 業種分散の最終形ポートフォリオ設計