Advanced 研究ワークブック

研究 ID / 著者 / 日付 / バージョン：
主な問いと仕組み：誰が払い、どのリスクを負うか？
Observation / Mechanism / Hypothesis：
帰無仮説 H0 / 反証条件 / 経済的意味の最低水準：
研究単位 / 保有期間 / 重複の処理：
データ源 / 取得時刻 / 最初に見えた時刻 / バージョン / 許諾：
資産ユニバース / 上場廃止 / 欠損 / ラベルの改訂：
学習・検証・最終テストの開始点と終了点：
すべての試行と失敗 / パラメーター選択規則：
ベースライン / アブレーション / コストストレス：
純収益分布 / 標本数 / 区間 / テール：
時系列依存 / 多重試行 / 未対応の仕組み：
結論：支持、棄却、証拠不足。具体的な出力を参照：
次の行動：研究継続、Testing、一時停止、退役。責任者：
入力ファイル / コードバージョン / 実行コマンド / 出力検証：

Strategy ID: 安定した一意の識別子
Name: 読める名称とバージョン
Market: 市場、取引場所、タイムゾーン
Instrument: 対象、契約仕様、通貨、各レッグ
Horizon: シグナル頻度、保有上限、ラベル区間
Category: 収益の仕組みによる分類。適用場面も明記
Hypothesis: 否定可能な主張と結果指標
Mechanism: 参加者、制約、補償が存在する理由
Data: 出所、可視時刻、バージョン、欠損・改訂処理
Features: 過去だけを使う計算窓、変換、単位
Signal: 明確な論理、閾値、無効入力時の動作
Entry: 提出可能時刻、実行可能価格、未約定処理
Exit: 通常・時間・失効による退出と、状態不明の注文
Sizing: 目標名目額、リスクによる縮尺、厳格な上限
Risk: 単一戦略、組合せ、場所、融資、停止の制限
Execution: 注文方式、子注文、期限、遅延、残数量
Cost: Fee、Slippage、融資などの内訳。重複禁止
Backtest: 入力、コード、基準、コスト、完全な出力
OOS: 隔離したテスト境界、漏洩確認、失敗標本
Regime: 当時分かる状態、適用範囲、既定の動作
Failure: 仕組みを棄却する証拠と、停止する条件
Capacity: 通常・ストレス時の容量、資本、退出制約
Status: ライフサイクルの状態、承認者、日付、証拠

オブジェクト ID / バージョン / 対応 Memo：
データ一覧と検証値 / 時刻と可視性：
資産ユニバースの履歴 / 欠損・上場廃止の処理：
規則、パラメーター、シード、全試行数：
約定・資金の仮定 / 初期資本 / 費用：
学習・検証・テスト境界 / Purge・Embargo の根拠：
総収益、純収益、最大ドローダウン、回転率、エクスポージャー：
現金・保有の基準 / 単一シグナルのアブレーション：
パラメーター近傍 / 状態 / コスト / 容量 / ギャップストレス：
各期の純資産 / 約定 / 未約定 / 拒否 / 現金移動：
統計区間と依存の仮定 / 対応していないリスク：
再計算手順 / 期待結果 / 失敗時の停止条件：
結論と昇格判断。独立した評価者の意見も含める：

報告時刻 / データ締切 / 戦略バージョン / 評価通貨：
総資本 / 配分済み / 現金 / 利用可能証拠金：
各戦略の純・総エクスポージャー、リスク予算、取引場所：
因子の定義 / Beta / その他の因子 / 非線形エクスポージャー：
共分散の窓 / 相関行列 / リスク寄与とその合計：
ボラティリティ目標 / 実際の変動 / レバレッジ・回転率上限：
ストレス入力、損失、資金不足、退出可能量：
ドローダウン / 到達した段階 / 注文・保有の残存リスク：
集中度 / 相手先 / データと執行の共通依存：
不足事項 / 責任者 / 即時行動 / 次の確認：
承認・停止判断 / 再開条件 / 独立した承認者：

親注文 ID / 方向 / 数量 / 判断時刻 / 判断価格：
到達時刻と基準 / 最終終了時刻 / 終了時の評価価格：
3種類の執行規則と共通入力。事前に固定：
各子注文：時刻、場所、数量、約定、残数量、費用：
総約定 / 残数量 / 平均価格 / 到達価格との差 / 判断価格との差：
機会コスト / 完了時間 / 逆選択 / 処理中のリスク：
価格経路、深度、手数料、遅延を1変数ずつ比較：
順番、インパクト、ルーティングの仮定 / 未立証事項：
結論の適用範囲 / 規則の維持・変更を支持する証拠：

四半期と戦略 ID / 期初計画 / 今期の変更：
9項目の PnL Attribution / 独立帳簿 / Residual の調査：
シグナル、コスト、容量、相関、因子の変化：
仕組みがなお成立するか / 反例 / 失敗した試行：
リスク・規律の事象 / 停止・再開記録：
ライフサイクル判断：維持、降格、一時停止、退役：
次期の資本表 / リスク予算 / 研究優先順位：
市場構造の変化と検証可能な証拠：
提案者 / リスク側の異議 / 承認者 / 発効時刻：

"""合成データによる教学研究。Python 3.10 以降、標準ライブラリのみ。注文と通信は行わない。"""
import hashlib
import json
import math
import random
import statistics

SEED = 20260930
HORIZON = 2
LOOKBACK = 10
CANDIDATES = (0.0, 0.003, 0.006)


def validate(returns):
    if len(returns) < 300 or any(not math.isfinite(x) or x <= -1 for x in returns):
        raise ValueError("-1 を超える有限の単純収益率が 300 件以上必要です")


def data(seed=SEED):
    rng = random.Random(seed)
    # 独立した合成収益率であり、市場の予測可能性を主張するものではない。
    return [rng.gauss(0, 0.015) for _ in range(300)]


def rows(returns):
    validate(returns)
    return [dict(t=t, end=t + HORIZON,
                 feature=sum(returns[t-LOOKBACK+1:t+1]),
                 label=math.prod(1+x for x in returns[t+1:t+HORIZON+1])-1)
            for t in range(LOOKBACK-1, len(returns)-HORIZON, HORIZON)]


def segment(samples, start, stop):
    # 右側の境界をまたぐラベルを除去する。特徴量は過去だけを使う。
    return [r for r in samples if start <= r["t"] < stop and r["end"] < stop]


def evaluate(samples, threshold, cost_bps=10):
    if not samples or not math.isfinite(cost_bps) or cost_bps < 0:
        raise ValueError("空でない標本と、非負の有限コストが必要です")
    if not math.isfinite(threshold) or threshold < 0:
        raise ValueError("閾値は有限かつ非負である必要があります")
    pnl = []
    traded = 0
    for r in samples:
        # 合成周期には寄り付きのギャップがなく、次の始値は t の終値と等しい。
        # t の終値後に判断し、次の始値で入り、2周期の収益率を保有する。
        position = int(r["feature"] > threshold)
        # 往復の比例コスト。レバレッジ・融資・重複する取引はない。
        pnl.append(position * r["label"] - 2 * cost_bps / 10000 * position)
        traded += position
    return dict(mean=statistics.mean(pnl), traded=traded, returns=pnl)


def choose(samples):
    # 同点は登録順の最初の候補を選ぶ。テスト入力は使わない。
    return max(CANDIDATES, key=lambda p: evaluate(samples, p)["mean"])


def max_drawdown(returns):
    equity = peak = 1.0
    worst = 0.0
    for r in returns:
        if not math.isfinite(r) or r <= -1:
            raise ValueError("経路の収益率が無効です")
        equity *= 1+r
        peak = max(peak, equity)
        worst = max(worst, 1-equity/peak)
    return worst


def block_bootstrap(returns, block=3, repetitions=1000, seed=SEED):
    if not returns or not isinstance(block, int) or not 1 <= block <= len(returns):
        raise ValueError("ブロックが無効です")
    if not isinstance(repetitions, int) or repetitions < 20:
        raise ValueError("少なくとも 20 回の反復が必要です")
    rng = random.Random(seed)
    means, drawdowns = [], []
    for _ in range(repetitions):
        resample = []
        while len(resample) < len(returns):
            start = rng.randrange(len(returns)-block+1)
            resample.extend(returns[start:start+block])
        resample = resample[:len(returns)]
        means.append(statistics.mean(resample))
        drawdowns.append(max_drawdown(resample))
    means.sort()
    drawdowns.sort()
    # 経験的な最近接インデックスの百分位であり、解析的な信頼性保証ではない。
    quantile = lambda xs, p: xs[round((len(xs)-1)*p)]
    return dict(mean_interval_95=[quantile(means, .025), quantile(means, .975)],
                drawdown_p50=quantile(drawdowns, .5),
                drawdown_p95=quantile(drawdowns, .95), block=block,
                repetitions=repetitions, seed=seed)


def run(returns):
    samples = rows(returns)
    folds = []
    for boundary in (100, 140, 180):
        training = segment(samples, 0, boundary)
        test = segment(samples, boundary, boundary+40)
        threshold = choose(training)
        folds.append(dict(train_end=boundary, test_end=boundary+40,
                          threshold=threshold, test_mean=evaluate(test, threshold)["mean"]))
    # 開発は 220 で終了。180..220 の検証で選び、最終テストは見ない。
    validation = segment(samples, 180, 220)
    selected = choose(validation)
    final = segment(samples, 220, len(returns))
    base = evaluate(final, selected)
    stressed = evaluate(final, selected, cost_bps=30)
    return dict(seed=SEED, input_sha256=hashlib.sha256(json.dumps(returns).encode()).hexdigest(),
                input_rows=len(returns), horizon=HORIZON, attempts=list(CANDIDATES),
                folds=folds, final_start=220, selected=selected, observations=len(final),
                traded=base["traded"], test_net_mean=base["mean"],
                cash_mean=0.0, always_long_round_trip_block_mean=statistics.mean(r["label"] for r in final)-0.002,
                high_cost_mean=stressed["mean"], max_drawdown=max_drawdown(base["returns"]),
                uncertainty=block_bootstrap(base["returns"]))


def self_test():
    original = data()
    samples = rows(original)
    for boundary in (100, 140, 180, 220):
        training = segment(samples, 0, boundary)
        assert all(r["end"] < boundary for r in training)
        assert any(r["t"] < boundary <= r["end"] for r in samples)
    changed = original[:220] + [0.1] * (len(original)-220)
    assert segment(rows(changed), 0, 220) == segment(samples, 0, 220)
    assert run(changed)["selected"] == run(original)["selected"]
    assert all(a["end"] <= b["t"] for a, b in zip(samples, samples[1:]))
    result = run(original)
    assert result == run(original)
    assert result["high_cost_mean"] <= result["test_net_mean"]
    assert abs(max_drawdown([.1, -.2, .1]) - .2) < 1e-12
    for invalid in ([], [0.0]*299, [math.nan]*300, [-1.0]*300):
        try:
            rows(invalid)
        except ValueError:
            pass
        else:
            raise AssertionError("無効なデータを受理しました")
    for block in (0, 1000):
        try:
            block_bootstrap([.01, -.01], block=block)
        except ValueError:
            pass
        else:
            raise AssertionError("無効なブロックを受理しました")


if __name__ == "__main__":
    self_test()
    print(json.dumps(run(data()), ensure_ascii=False, indent=2))