📌 はじめに
株式や先物のトレード戦略を作る際、過剰最適化を避けるために「ウォークフォワード分析（WFA）」が有効です。
今回は、Pythonのフレームワーク FastAPI と最適化ライブラリ Optuna を使って、
誰でもブラウザからウォークフォワード検証ができるWebアプリを開発しました。
⚙️ 使用技術
- Python 3.8
- FastAPI
- Optuna
- Matplotlib
- pandas, numpy 他
🖥️ デモ（Renderで一時公開中）
Render:https://my-flask-app-mksr.onrender.com/
🧠 ウォークフォワード分析とは？
過去データを「学習期間」と「検証期間」に分けて、順次ずらしながら最適化と評価を行う手法。
過剰最適化（カーブフィッティング）を避けやすく、実運用に近い評価が可能です。
バックテストでk1,k2の値を試行回数分最適化して，フォワードテストで最適化されたk1,k2の値を使って評価をする，という流れ
🧪 特徴と機能
- ブラウザ上で戦略のパラメータを入力
- 処理ログとリアルタイム出力
- バックテスト後に画像と結果を表示
- CSVへの結果保存
- Renderによるスマホアクセス対応
Noteにもシステムの特徴をざっくり紹介しています。参考まで。
https://note.com/trade_python/n/ne4d70a11c087
検証結果
- Randomでのエントリー ＆ Randomエグジット
- Randomでのエントリー ＆ ダウ理論(トレーリングストップ)でのエグジット
- Golden crossでのエントリー ＆ エグジット
などの検証結果をブログにてまとめてます
ウォークフォワード分析とは？Pythonで戦略を自動評価するWebアプリを作成
🔧 コード抜粋（FastAPI）
from fastapi import FastAPI, Request, Form
from fastapi.responses import HTMLResponse, StreamingResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
app = FastAPI()
templates = Jinja2Templates(directory="templates")
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
return templates.TemplateResponse("index.html", {
"request": request,
"log": None,
"img_base64": None,
"knum": None,
"TrialNum": None,
"condition": "",
"condition1": ""
})
@app.get("/get_image", response_class=JSONResponse)
async def get_image():
try:
with open("result/last_img.png", "r") as f:
base64_img = f.read()
return {"img": base64_img}
except:
return {"img": ""}
@app.post("/run")
async def run(knum: int = Form(...), TrialNum: int = Form(...),
condition: str = Form(""), condition1: str = Form("")):
def generate():
## 🔧 コード抜粋（Optunaによるパラメータ最適化）
```python
def objective(trial,t0,t1):
if knum>0:
k1= trial.suggest_int('k1',2,35)
# k1= trial.suggest_int('k1', 0,23)
# k1= trial.suggest_float('k1', 0,1)
k2=0
k3=0
k4=0
if knum>1:
k2= trial.suggest_int('k2', 2,20)
# k2= trial.suggest_int('k2', 10,20)
# k2= trial.suggest_float('k2',0,1)
k3=0
k4=0
if knum>2:
# k3= trial.suggest_int('k3', 20,80)
k3= trial.suggest_float('k3', 0,1)
k4=0
if knum>3:
# k4= trial.suggest_int('k4', 1,5)
k4= trial.suggest_float('k4', 0,1)
# histry=1.0
# histry1=backtest(k1,k2,t0,t1,"L",0)
histry1=backtest(k1,k2,k3,k4,t0,t1,LSt,0,finishList0,rate,Close,Open,Low,High,Vol,condition,condition1)
return histry1
step0=2400
aaa=1750
if hd=="d":
step0=100
aaa=210
for lp in range(step0*5+aaa,len(Close)-step0-10,step0):
log_stream = io.StringIO()
sys.stdout = log_stream # 標準出力をキャプチャ
# t0=aaa
t0=lp-step0*5
t1=lp
t2=lp+step0+2
prog = int(lp/len(Close)*100)
try:
optuna.logging.disable_default_handler()
study = optuna.create_study(direction='maximize', pruner=SuccessiveHalvingPruner())
study.optimize(lambda trial: objective(trial,t0,t1), n_trials=TrialNum)
print(f"progress {cur},{prog}%")
##print(f"The best parameters are : \n {study.best_params}")
except Exception as e:
print(f"error {e}")
finally:
sys.stdout = sys.__stdout__ # 出力先を戻す
for line in log_stream.getvalue().splitlines():
yield f"{line}\n".encode("utf-8")
if knum==1:
k1 = study.best_params["k1"]
k2=0
k3=0
k4=0
histry1=backtest(k1,k2,k3,k4,t1,t2,LSt,1,finishList0,rate,Close,Open,Low,High,Vol,condition,condition1)
# print("best setting",[k1])
if knum==2:
k1 = study.best_params["k1"]
k2 = study.best_params["k2"]
k3=0
k4=0
histry1=backtest(k1,k2,k3,k4,t1,t2,LSt,1,finishList0,rate,Close,Open,Low,High,Vol,condition,condition1)
# print("best setting",[k1,k2])