熊本で震度７の地震がありました。
今年、西南戦争を巡る旅として熊本に旅行に行ったばかりなので、他人事とは思えず、大変驚いています。復旧作業中だった熊本城の石垣も一部崩れてしまったとのことで悲しい気持ちです。
熊本でお世話になった郷土料理のお店や地酒のお店の人たちも無事でありますように。
.
.
前回は曜日別の損益を分析しました。
その結果、利益の多くは2～4時間以内に生まれており、4時間を超えて保有した取引では成績が悪化する傾向が見られました。
今回は、取引の連勝・連敗を分析します。
勝率や総損益だけでなく、「最大で何回くらい連敗するのか」も大事な要素です。
あらかじめ過去の最大連敗数を把握しておけば、実際に連敗した際に、通常の範囲内なのかを判断する材料になります。
今回調べること
今回は、次の内容をPythonで確認します。
・最大連勝数
・最大連敗数
・連勝、連敗の発生回数
・連勝数、連敗数ごとの分布
・それぞれの連勝、連敗期間に発生した損益
今回は保有時間別に分析してみます。
前回に作成したPythonファイル「analyze.py」に以下のコードを追記します。
# ==============================
# 連勝・連敗分析
# ==============================
# 決済時間の順番に並べる
streak_trades = (
holding_trades
.sort_values("決済時間")
.reset_index(drop=True)
.copy()
)
# 各取引を勝ち・負け・引き分けに分類
def classify_result(profit):
if profit > 0:
return "勝ち"
elif profit < 0:
return "負け"
else:
return "引き分け"
streak_trades["結果"] = (
streak_trades["損益"]
.apply(classify_result)
)
# 取引順を追加
streak_trades["取引順"] = range(
1,
len(streak_trades) + 1
)
# 直前の取引と結果が違う場所を判定
result_changed = (
streak_trades["結果"]
!= streak_trades["結果"].shift()
)
# 結果が切り替わるたびにグループ番号を付ける
streak_trades["連続グループ"] = (
result_changed.cumsum()
)
# 各連勝・連敗グループを集計
streak_groups = (
streak_trades
.groupby("連続グループ")
.agg(
結果=("結果", "first"),
連続回数=("結果", "size"),
開始取引=("取引順", "min"),
終了取引=("取引順", "max"),
開始時間=("決済時間", "min"),
終了時間=("決済時間", "max"),
期間損益=("損益", "sum"),
平均損益=("損益", "mean")
)
.reset_index(drop=True)
)
# 勝ちと負けだけを対象にする
streak_groups = streak_groups[
streak_groups["結果"].isin(["勝ち", "負け"])
].copy()
# 連勝データと連敗データに分ける
winning_streaks = streak_groups[
streak_groups["結果"] == "勝ち"
].copy()
losing_streaks = streak_groups[
streak_groups["結果"] == "負け"
].copy()
# 基本情報を計算
max_winning_streak = (
winning_streaks["連続回数"].max()
if not winning_streaks.empty
else 0
)
max_losing_streak = (
losing_streaks["連続回数"].max()
if not losing_streaks.empty
else 0
)
average_winning_streak = (
winning_streaks["連続回数"].mean()
if not winning_streaks.empty
else 0
)
average_losing_streak = (
losing_streaks["連続回数"].mean()
if not losing_streaks.empty
else 0
)
# 最大連勝・最大連敗に該当する期間
max_winning_streak_rows = winning_streaks[
winning_streaks["連続回数"]
== max_winning_streak
].copy()
max_losing_streak_rows = losing_streaks[
losing_streaks["連続回数"]
== max_losing_streak
].copy()
# 結果を表示
print("\n===== 連勝・連敗分析 =====")
print(f"最大連勝数: {max_winning_streak}回")
print(f"最大連敗数: {max_losing_streak}回")
print(f"平均連勝数: {average_winning_streak:.2f}回")
print(f"平均連敗数: {average_losing_streak:.2f}回")
print(f"連勝の発生回数: {len(winning_streaks)}回")
print(f"連敗の発生回数: {len(losing_streaks)}回")
print("\n===== 最大連勝期間 =====")
print(
max_winning_streak_rows[
[
"連続回数",
"開始時間",
"終了時間",
"期間損益"
]
].round(2)
)
print("\n===== 最大連敗期間 =====")
print(
max_losing_streak_rows[
[
"連続回数",
"開始時間",
"終了時間",
"期間損益"
]
].round(2)
)
# 連勝数別の分布
winning_streak_distribution = (
winning_streaks["連続回数"]
.value_counts()
.sort_index()
.rename_axis("連勝数")
.reset_index(name="発生回数")
)
# 連敗数別の分布
losing_streak_distribution = (
losing_streaks["連続回数"]
.value_counts()
.sort_index()
.rename_axis("連敗数")
.reset_index(name="発生回数")
)
print("\n===== 連勝数ごとの発生回数 =====")
print(winning_streak_distribution)
print("\n===== 連敗数ごとの発生回数 =====")
print(losing_streak_distribution)
# CSVに保存
streak_trades.to_csv(
"results/streak_trades.csv",
index=False,
encoding="utf-8-sig"
)
streak_groups.to_csv(
"results/streak_groups.csv",
index=False,
encoding="utf-8-sig"
)
winning_streak_distribution.to_csv(
"results/winning_streak_distribution.csv",
index=False,
encoding="utf-8-sig"
)
losing_streak_distribution.to_csv(
"results/losing_streak_distribution.csv",
index=False,
encoding="utf-8-sig"
)
print(
"\n連勝・連敗分析結果を"
"resultsフォルダに保存しました。"
)
# 連勝数の分布グラフ
plt.figure(figsize=(10, 6))
plt.bar(
winning_streak_distribution["連勝数"],
winning_streak_distribution["発生回数"]
)
plt.title("Winning Streak Distribution")
plt.xlabel("Number of Consecutive Wins")
plt.ylabel("Frequency")
plt.xticks(
winning_streak_distribution["連勝数"]
)
plt.tight_layout()
plt.savefig(
"results/winning_streak_distribution.png",
dpi=150
)
plt.close()
# 連敗数の分布グラフ
plt.figure(figsize=(10, 6))
plt.bar(
losing_streak_distribution["連敗数"],
losing_streak_distribution["発生回数"]
)
plt.title("Losing Streak Distribution")
plt.xlabel("Number of Consecutive Losses")
plt.ylabel("Frequency")
plt.xticks(
losing_streak_distribution["連敗数"]
)
plt.tight_layout()
plt.savefig(
"results/losing_streak_distribution.png",
dpi=150
)
plt.close()
print(
"連勝・連敗分布のグラフを"
"resultsフォルダに保存しました。"
)コードのポイント
sort_values()で決済順に並べる
.sort_values("決済時間")連勝・連敗を調べるためには、取引を実際に決済された順番に並べる必要があります。
CSVの行順が必ずしも取引の発生順とは限らないため、最初に決済時間で並べ替えています。
関数を作って取引結果を分類する
def classify_result(profit):
if profit > 0:
return "勝ち"
elif profit < 0:
return "負け"
else:
return "引き分け"損益がプラスなら「勝ち」、マイナスなら「負け」、0なら「引き分け」と分類しています。
その後、.apply()を使い、すべての取引にこの関数を適用しています。
streak_trades["結果"] = (
streak_trades["損益"]
.apply(classify_result)
).shift()で直前の取引と比較する
streak_trades["結果"].shift().shift()は、データを1行ずらす処理です。
これによって、現在の取引結果と一つ前の取引結果を比較できます。
例えば、結果が次のようになっていたとします。
勝ち
勝ち
負け
負け
勝ち直前の行と結果が異なる場所を判定すると、次のようになります。
True
False
True
False
True最初の勝ち、最初の負け、次の勝ちの部分で、新しいグループが始まっていることが分かります。
.cumsum()でグループ番号を作る
result_changed.cumsum().cumsum()は累積合計を求めるメソッドです。
Trueは1、Falseは0として計算されるため、結果が切り替わるたびに番号が増えます。
勝ち グループ1
勝ち グループ1
負け グループ2
負け グループ2
勝ち グループ3このグループごとの取引数を数えることで、連勝数と連敗数を求めています。
出力した結果はこちらです。
分析
分析の結果、「零号機」は1～2連勝・1～2連敗で終わるケースが大半を占め、連続回数が増えるにつれて発生頻度は急激に減少しました。
最大連勝は11回、最大連敗は9回であり、長期間連敗が続くケースは極めて少なかったですが、5連勝以上は比較的多く確認でき、勝ち始めると利益を伸ばせる傾向も見らます。
9連敗が2回発生しています。バックテストの最大連敗は将来の上限ではありませんが、10～12連敗以上が発生した場合は何らからの不具合や環境の変化を疑う必要がありそうです。
小さな勝ちを何度も積み重ねるタイプということがわかりました。
今日はここまで。
次回は、資産推移からドローダウンを計算し、「零号機」が過去にどの程度の下落を経験したのかを詳しく分析したいと思います。
おまけ
今、この本を読んでいます。
ユヴァル・ノア・ハラリの「NEXUS」を詠んだ後、同じ雰囲気の本ということで、興味があり手に取ってみました。
大陸や民族によって、こうも文明文化の発展度合いが違うのはなぜなのか。
その答えを探すべく、人類誕生の700万年前からさかのぼり答えを探すというものです。
スペインのピサロがインカ帝国アタワルパを捕らえ、処刑した事実は知っていましたが、実際にどんな感じだったのか興味はありましたがずっと知らないままでした。
この本では、ピサロの下で働いていた部下が報告書で話している内容を引用することで、どのように占拠したのかが描写されています。
まだまだ途中ですが続きが気になる本です。
コメント