ランダムウォークへの挑戦⑳｜零号機の利益は「一部の大勝ち」に支えられているのか（2026/8/20）

import os
import math
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt


# ============================================================
# 設定
# ============================================================

INPUT_FILE = "results/trade_base.csv"

OUTPUT_DIR = "results"

os.makedirs(
    OUTPUT_DIR,
    exist_ok=True
)


OUTPUT_SUMMARY_CSV = os.path.join(
    OUTPUT_DIR,
    "profit_distribution_analysis.csv"
)

OUTPUT_CONTRIBUTION_CSV = os.path.join(
    OUTPUT_DIR,
    "profit_contribution_analysis.csv"
)

OUTPUT_STRESS_CSV = os.path.join(
    OUTPUT_DIR,
    "profit_stress_test.csv"
)

OUTPUT_BOXPLOT = os.path.join(
    OUTPUT_DIR,
    "profit_distribution_boxplot.png"
)

OUTPUT_BOXPLOT_TRIMMED = os.path.join(
    OUTPUT_DIR,
    "profit_distribution_boxplot_trimmed.png"
)

OUTPUT_CONTRIBUTION_GRAPH = os.path.join(
    OUTPUT_DIR,
    "top_profit_contribution.png"
)

OUTPUT_STRESS_GRAPH = os.path.join(
    OUTPUT_DIR,
    "profit_factor_stress_test.png"
)


# ============================================================
# データ読み込み
# ============================================================

df = pd.read_csv(
    INPUT_FILE
)


print(
    "========================================"
)

print(
    "PROFIT DISTRIBUTION ANALYSIS"
)

print(
    "========================================"
)

print(
    f"\n読み込み件数: {len(df)}"
)


# ============================================================
# 必要列確認
# ============================================================

required_columns = [
    "entry_band_distance_pips",
    "entry_atr14_pips",
    "profit"
]


missing_columns = [
    column
    for column in required_columns
    if column not in df.columns
]


if missing_columns:

    raise ValueError(
        "必要な列がありません: "
        + ", ".join(missing_columns)
    )


# ============================================================
# 数値型へ変換
# ============================================================

for column in required_columns:

    df[column] = pd.to_numeric(
        df[column],
        errors="coerce"
    )


# ============================================================
# 分析対象
# ============================================================

analysis_df = df[
    df["entry_band_distance_pips"].notna()
    &
    df["entry_atr14_pips"].notna()
    &
    df["profit"].notna()
    &
    (
        df["entry_atr14_pips"]
        > 0
    )
].copy()


if analysis_df.empty:

    raise ValueError(
        "分析対象データがありません。"
    )


print(
    f"分析対象件数: {len(analysis_df)}"
)


# ============================================================
# Relative Band Distance
# ============================================================

analysis_df[
    "relative_band_distance"
] = (
    analysis_df[
        "entry_band_distance_pips"
    ]
    /
    analysis_df[
        "entry_atr14_pips"
    ]
)


analysis_df[
    "relative_band_distance"
] = (
    analysis_df[
        "relative_band_distance"
    ]
    .replace(
        [
            np.inf,
            -np.inf
        ],
        np.nan
    )
)


analysis_df = analysis_df[
    analysis_df[
        "relative_band_distance"
    ]
    .notna()
].copy()


# ============================================================
# Relative Band Distanceを四分位化
# ============================================================

quartile_order = [
    "Q1",
    "Q2",
    "Q3",
    "Q4"
]


analysis_df[
    "relative_band_group"
] = pd.qcut(
    analysis_df[
        "relative_band_distance"
    ],
    q=4,
    labels=quartile_order
)


# ============================================================
# PF計算関数
# ============================================================

def calculate_pf(data):

    gross_profit = (
        data.loc[
            data["profit"] > 0,
            "profit"
        ]
        .sum()
    )

    gross_loss = abs(
        data.loc[
            data["profit"] < 0,
            "profit"
        ]
        .sum()
    )

    if gross_loss == 0:

        return np.nan

    return (
        gross_profit
        /
        gross_loss
    )


# ============================================================
# 分布統計
# ============================================================

summary_rows = []


for group_name in quartile_order:

    group_df = analysis_df[
        analysis_df[
            "relative_band_group"
        ]
        == group_name
    ].copy()


    profit = group_df[
        "profit"
    ]


    summary_rows.append(
        {
            "group": group_name,

            "trades": len(
                group_df
            ),

            "wins": (
                profit > 0
            ).sum(),

            "losses": (
                profit < 0
            ).sum(),

            "win_rate": (
                (
                    profit > 0
                ).mean()
                * 100
            ),

            "total_profit": (
                profit.sum()
            ),

            "average_profit": (
                profit.mean()
            ),

            "minimum_profit": (
                profit.min()
            ),

            "q25_profit": (
                profit.quantile(
                    0.25
                )
            ),

            "median_profit": (
                profit.median()
            ),

            "q75_profit": (
                profit.quantile(
                    0.75
                )
            ),

            "q90_profit": (
                profit.quantile(
                    0.90
                )
            ),

            "q95_profit": (
                profit.quantile(
                    0.95
                )
            ),

            "maximum_profit": (
                profit.max()
            ),

            "profit_factor": (
                calculate_pf(
                    group_df
                )
            )
        }
    )


summary_df = pd.DataFrame(
    summary_rows
)


summary_df = (
    summary_df
    .set_index(
        "group"
    )
)


# ============================================================
# 小数点整理
# ============================================================

summary_df[
    "win_rate"
] = (
    summary_df[
        "win_rate"
    ]
    .round(2)
)


round_columns = [
    "total_profit",
    "average_profit",
    "minimum_profit",
    "q25_profit",
    "median_profit",
    "q75_profit",
    "q90_profit",
    "q95_profit",
    "maximum_profit"
]


for column in round_columns:

    summary_df[
        column
    ] = (
        summary_df[
            column
        ]
        .round(2)
    )


summary_df[
    "profit_factor"
] = (
    summary_df[
        "profit_factor"
    ]
    .round(3)
)


print(
    "\n========================================"
)

print(
    "損益分布"
)

print(
    "========================================"
)


print(
    summary_df
)


summary_df.to_csv(
    OUTPUT_SUMMARY_CSV,
    encoding="utf-8-sig"
)


# ============================================================
# 上位利益寄与率
# ============================================================

contribution_rows = []


for group_name in quartile_order:

    group_df = analysis_df[
        analysis_df[
            "relative_band_group"
        ]
        == group_name
    ].copy()


    winning_df = group_df[
        group_df[
            "profit"
        ]
        > 0
    ].copy()


    winning_df = winning_df.sort_values(
        by="profit",
        ascending=False
    )


    gross_profit = (
        winning_df[
            "profit"
        ]
        .sum()
    )


    win_count = len(
        winning_df
    )


    top_10_count = max(
        1,
        math.ceil(
            win_count
            * 0.10
        )
    )


    top_5_count = max(
        1,
        math.ceil(
            win_count
            * 0.05
        )
    )


    top_10_profit = (
        winning_df
        .head(
            top_10_count
        )
        [
            "profit"
        ]
        .sum()
    )


    top_5_profit = (
        winning_df
        .head(
            top_5_count
        )
        [
            "profit"
        ]
        .sum()
    )


    if gross_profit > 0:

        top_10_share = (
            top_10_profit
            /
            gross_profit
            * 100
        )

        top_5_share = (
            top_5_profit
            /
            gross_profit
            * 100
        )

    else:

        top_10_share = np.nan
        top_5_share = np.nan


    contribution_rows.append(
        {
            "group": group_name,

            "winning_trades": win_count,

            "gross_profit": gross_profit,

            "top_10_count": top_10_count,

            "top_10_profit": top_10_profit,

            "top_10_profit_share": top_10_share,

            "top_5_count": top_5_count,

            "top_5_profit": top_5_profit,

            "top_5_profit_share": top_5_share
        }
    )


contribution_df = pd.DataFrame(
    contribution_rows
)


contribution_df = (
    contribution_df
    .set_index(
        "group"
    )
)


for column in [
    "gross_profit",
    "top_10_profit",
    "top_5_profit"
]:

    contribution_df[
        column
    ] = (
        contribution_df[
            column
        ]
        .round(2)
    )


for column in [
    "top_10_profit_share",
    "top_5_profit_share"
]:

    contribution_df[
        column
    ] = (
        contribution_df[
            column
        ]
        .round(2)
    )


print(
    "\n========================================"
)

print(
    "上位利益取引の寄与率"
)

print(
    "========================================"
)


print(
    contribution_df
)


contribution_df.to_csv(
    OUTPUT_CONTRIBUTION_CSV,
    encoding="utf-8-sig"
)


# ============================================================
# 上位勝ち取引除外ストレステスト
# ============================================================

stress_rows = []


for group_name in quartile_order:

    group_df = analysis_df[
        analysis_df[
            "relative_band_group"
        ]
        == group_name
    ].copy()


    original_pf = (
        calculate_pf(
            group_df
        )
    )


    winning_df = group_df[
        group_df[
            "profit"
        ]
        > 0
    ].sort_values(
        by="profit",
        ascending=False
    )


    win_count = len(
        winning_df
    )


    top_10_count = max(
        1,
        math.ceil(
            win_count
            * 0.10
        )
    )


    top_5_count = max(
        1,
        math.ceil(
            win_count
            * 0.05
        )
    )


    top_10_indices = (
        winning_df
        .head(
            top_10_count
        )
        .index
    )


    top_5_indices = (
        winning_df
        .head(
            top_5_count
        )
        .index
    )


    without_top_10 = (
        group_df
        .drop(
            index=top_10_indices
        )
    )


    without_top_5 = (
        group_df
        .drop(
            index=top_5_indices
        )
    )


    pf_without_top_10 = (
        calculate_pf(
            without_top_10
        )
    )


    pf_without_top_5 = (
        calculate_pf(
            without_top_5
        )
    )


    total_profit_without_top_10 = (
        without_top_10[
            "profit"
        ]
        .sum()
    )


    total_profit_without_top_5 = (
        without_top_5[
            "profit"
        ]
        .sum()
    )


    stress_rows.append(
        {
            "group": group_name,

            "original_pf": (
                original_pf
            ),

            "pf_without_top_5_percent_winners": (
                pf_without_top_5
            ),

            "pf_without_top_10_percent_winners": (
                pf_without_top_10
            ),

            "total_profit_without_top_5_percent_winners": (
                total_profit_without_top_5
            ),

            "total_profit_without_top_10_percent_winners": (
                total_profit_without_top_10
            )
        }
    )


stress_df = pd.DataFrame(
    stress_rows
)


stress_df = (
    stress_df
    .set_index(
        "group"
    )
)


for column in [
    "original_pf",
    "pf_without_top_5_percent_winners",
    "pf_without_top_10_percent_winners"
]:

    stress_df[
        column
    ] = (
        stress_df[
            column
        ]
        .round(3)
    )


for column in [
    "total_profit_without_top_5_percent_winners",
    "total_profit_without_top_10_percent_winners"
]:

    stress_df[
        column
    ] = (
        stress_df[
            column
        ]
        .round(2)
    )


print(
    "\n========================================"
)

print(
    "大勝ち除外ストレステスト"
)

print(
    "========================================"
)


print(
    stress_df
)


stress_df.to_csv(
    OUTPUT_STRESS_CSV,
    encoding="utf-8-sig"
)


# ============================================================
# 箱ひげ図
# ============================================================

boxplot_data = [
    analysis_df.loc[
        analysis_df[
            "relative_band_group"
        ]
        == group,
        "profit"
    ]
    for group in quartile_order
]


fig, ax = plt.subplots(
    figsize=(9, 6)
)


ax.boxplot(
    boxplot_data,
    tick_labels=quartile_order
)


ax.axhline(
    y=0,
    linestyle="--"
)


ax.set_xlabel(
    "Relative Band Distance Quartile"
)


ax.set_ylabel(
    "Profit"
)


ax.set_title(
    "Profit Distribution by Relative Band Distance"
)


ax.grid(
    axis="y",
    alpha=0.3
)


plt.tight_layout()


plt.savefig(
    OUTPUT_BOXPLOT,
    dpi=150
)


plt.close()


# ============================================================
# 上下1％を除いた箱ひげ図
# ============================================================

lower_limit = (
    analysis_df[
        "profit"
    ]
    .quantile(
        0.01
    )
)


upper_limit = (
    analysis_df[
        "profit"
    ]
    .quantile(
        0.99
    )
)


trimmed_df = analysis_df[
    (
        analysis_df[
            "profit"
        ]
        >= lower_limit
    )
    &
    (
        analysis_df[
            "profit"
        ]
        <= upper_limit
    )
].copy()


trimmed_boxplot_data = [
    trimmed_df.loc[
        trimmed_df[
            "relative_band_group"
        ]
        == group,
        "profit"
    ]
    for group in quartile_order
]


fig, ax = plt.subplots(
    figsize=(9, 6)
)


ax.boxplot(
    trimmed_boxplot_data,
    tick_labels=quartile_order
)


ax.axhline(
    y=0,
    linestyle="--"
)


ax.set_xlabel(
    "Relative Band Distance Quartile"
)


ax.set_ylabel(
    "Profit"
)


ax.set_title(
    "Profit Distribution by Relative Band Distance (1%-99%)"
)


ax.grid(
    axis="y",
    alpha=0.3
)


plt.tight_layout()


plt.savefig(
    OUTPUT_BOXPLOT_TRIMMED,
    dpi=150
)


plt.close()


# ============================================================
# 上位利益寄与率グラフ
# ============================================================

contribution_plot_df = (
    contribution_df[
        [
            "top_5_profit_share",
            "top_10_profit_share"
        ]
    ]
)


fig, ax = plt.subplots(
    figsize=(9, 6)
)


contribution_plot_df.plot(
    kind="bar",
    ax=ax
)


ax.set_xlabel(
    "Relative Band Distance Quartile"
)


ax.set_ylabel(
    "Share of Gross Profit (%)"
)


ax.set_title(
    "Contribution of Top Winning Trades"
)


ax.tick_params(
    axis="x",
    rotation=0
)


ax.grid(
    axis="y",
    alpha=0.3
)


plt.tight_layout()


plt.savefig(
    OUTPUT_CONTRIBUTION_GRAPH,
    dpi=150
)


plt.close()


# ============================================================
# PFストレステストグラフ
# ============================================================

stress_plot_df = (
    stress_df[
        [
            "original_pf",
            "pf_without_top_5_percent_winners",
            "pf_without_top_10_percent_winners"
        ]
    ]
)


fig, ax = plt.subplots(
    figsize=(9, 6)
)


stress_plot_df.plot(
    kind="bar",
    ax=ax
)


ax.axhline(
    y=1.0,
    linestyle="--"
)


ax.set_xlabel(
    "Relative Band Distance Quartile"
)


ax.set_ylabel(
    "Profit Factor"
)


ax.set_title(
    "Profit Factor Stress Test"
)


ax.tick_params(
    axis="x",
    rotation=0
)


ax.grid(
    axis="y",
    alpha=0.3
)


plt.tight_layout()


plt.savefig(
    OUTPUT_STRESS_GRAPH,
    dpi=150
)


plt.close()


# ============================================================
# 全体結果
# ============================================================

print(
    "\n========================================"
)

print(
    "分析完了"
)

print(
    "========================================"
)


print(
    "\n出力ファイル:"
)


print(
    OUTPUT_SUMMARY_CSV
)


print(
    OUTPUT_CONTRIBUTION_CSV
)


print(
    OUTPUT_STRESS_CSV
)


print(
    OUTPUT_BOXPLOT
)


print(
    OUTPUT_BOXPLOT_TRIMMED
)


print(
    OUTPUT_CONTRIBUTION_GRAPH
)


print(
    OUTPUT_STRESS_GRAPH
)