ในโลกของการพัฒนาระบบเทรดและ AI Model การทำ Backtest เป็นเครื่องมือสำคัญในการประเมินประสิทธิภาพ แต่ปัญหาที่พบบ่อยที่สุดคือ Overfitting — ระบบที่ดูมีประสิทธิภาพสูงบนข้อมูลในอดีตแต่ล้มเหลวเมื่อนำไปใช้จริง
บทความนี้จะพาคุณเรียนรู้ Walk-forward Analysis ตั้งแต่พื้นฐานจนถึงเทคนิคขั้นสูง พร้อมโค้ด Python ที่ใช้งานได้จริง และวิธีใช้ AI จาก HolySheep AI ช่วยวินิจฉัย Overfitting อย่างมีประสิทธิภาพ
Overfitting คืออะไร และทำไมต้องกังวล
Overfitting เกิดขึ้นเมื่อโมเดลหรือระบบเทรด "จดจำ" ข้อมูล training แทนที่จะ "เรียนรู้" รูปแบบทั่วไป ผลลัพธ์คือ:
- Backtest Return สูงเกินจริง — อาจเห็น Sharpe Ratio 5-10 แต่พอ live กลับขาดทุน
- Parameter Sensitivity สูง — เปลี่ยนค่า parameter เล็กน้อยก็ทำให้ผลลัพธ์เปลี่ยนมาก
- Zero or Negative Out-of-Sample Performance — ผลตอบแทนในช่วงทดสอบจริง (out-of-sample) แย่กว่า baseline หรือขาดทุน
Walk-forward Analysis คืออะไร
Walk-forward Analysis เป็นวิธีการทดสอบที่แบ่งข้อมูลเป็น In-sample (Training) และ Out-of-sample (Testing) อย่างเป็นระบบ โดยเลื่อน window ไปข้างหน้าเรื่อยๆ ทำให้เห็นว่าโมเดลทำงานได้ดีแค่ไหนในช่วงที่ไม่เคยเห็น
หลักการสำคัญของ Walk-forward
# หลักการ Walk-forward Analysis
"""
┌─────────────────────────────────────────────────────────┐
│ In-Sample (Train) │ Out-of-Sample (Test) │ Future │
├─────────────────────────────────────────────────────────┤
│ ◄── Window 1 ──► │ │ │
│ │ ◄── Window 2 ──► │ │
│ │ │ ◄──► │
└─────────────────────────────────────────────────────────┘
Step 1: Train บนข้อมูล Window 1
Step 2: Test บนข้อมูล Window 2 (ที่ไม่เคยเห็น)
Step 3: เลื่อน Window ไปข้างหน้า ทำซ้ำ
"""
โค้ด Python สำหรับ Walk-forward Analysis
นี่คือโค้ดที่ใช้งานได้จริงสำหรับวินิจฉัย Overfitting ด้วย Walk-forward Analysis
import numpy as np
import pandas as pd
from sklearn.model_selection import TimeSeriesSplit
import matplotlib.pyplot as plt
from dataclasses import dataclass
from typing import List, Tuple, Dict
import warnings
warnings.filterwarnings('ignore')
@dataclass
class WalkForwardResult:
window: int
train_score: float
test_score: float
degradation: float # ความแตกต่างระหว่าง train กับ test
is_stable: bool
class OverfittingDiagnoser:
"""
เครื่องมือวินิจฉัย Overfitting ด้วย Walk-forward Analysis
"""
def __init__(self, n_splits: int = 10, test_size: float = 0.2):
self.n_splits = n_splits
self.test_size = test_size
self.results: List[WalkForwardResult] = []
def run_walkforward_analysis(
self,
X: np.ndarray,
y: np.ndarray,
model_generator: callable
) -> pd.DataFrame:
"""
รัน Walk-forward Analysis
Parameters:
-----------
X : np.ndarray - ข้อมูล Features
y : np.ndarray - ข้อมูล Target
model_generator : callable - ฟังก์ชันสร้างโมเดลใหม่
Returns:
--------
pd.DataFrame - ผลลัพธ์ของแต่ละ window
"""
tscv = TimeSeriesSplit(n_splits=self.n_splits)
all_results = []
print(f"เริ่ม Walk-forward Analysis ({self.n_splits} windows)")
print("=" * 50)
for window_idx, (train_idx, test_idx) in enumerate(tscv.split(X)):
# ตรวจสอบว่าข้อมูลเพียงพอหรือไม่
if len(train_idx) < 50 or len(test_idx) < 10:
continue
X_train, X_test = X[train_idx], X[test_idx]
y_train, y_test = y[train_idx], y[test_idx]
# สร้างและ train โมเดลใหม่ทุก window
model = model_generator()
model.fit(X_train, y_train)
# วัดผล
train_score = model.score(X_train, y_train)
test_score = model.score(X_test, y_test)
degradation = train_score - test_score
# ตรวจสอบความเสถียร (test score > 0, degradation < 0.3)
is_stable = test_score > 0 and degradation < 0.3
result = WalkForwardResult(
window=window_idx + 1,
train_score=train_score,
test_score=test_score,
degradation=degradation,
is_stable=is_stable
)
self.results.append(result)
print(f"Window {window_idx + 1}: "
f"Train={train_score:.4f}, "
f"Test={test_score:.4f}, "
f"Degradation={degradation:.4f}, "
f"Stable={'✓' if is_stable else '✗'}")
all_results.append({
'window': window_idx + 1,
'train_score': train_score,
'test_score': test_score,
'degradation': degradation,
'is_stable': is_stable
})
return pd.DataFrame(all_results)
def get_diagnosis_report(self) -> Dict:
"""
สร้างรายงานการวินิจฉัย Overfitting
"""
if not self.results:
return {"status": "No results", "recommendations": []}
df = pd.DataFrame([{
'train_score': r.train_score,
'test_score': r.test_score,
'degradation': r.degradation,
'is_stable': r.is_stable
} for r in self.results])
# คำนวณ metrics
avg_degradation = df['degradation'].mean()
stability_rate = df['is_stable'].mean()
test_score_std = df['test_score'].std()
# วินิจฉัย
if avg_degradation > 0.5:
diagnosis = "SEVERE_OVERFITTING"
severity = "สูงมาก"
color = "🔴"
elif avg_degradation > 0.3:
diagnosis = "MODERATE_OVERFITTING"
severity = "ปานกลาง"
color = "🟡"
elif avg_degradation > 0.1:
diagnosis = "MINOR_OVERFITTING"
severity = "เล็กน้อย"
color = "🟢"
else:
diagnosis = "HEALTHY_MODEL"
severity = "ปกติ"
color = "✅"
recommendations = []
if avg_degradation > 0.3:
recommendations.append("• ลดความซับซ้อนของโมเดล (ลด features หรือ parameters)")
recommendations.append("• เพิ่มข้อมูล training")
recommendations.append("• ใช้ Regularization เช่น L1/L2")
if test_score_std > 0.1:
recommendations.append("• พิจารณา ensemble methods")
if stability_rate < 0.7:
recommendations.append("• โมเดลไม่เสถียร ควรตรวจสอบ data quality")
return {
"diagnosis": diagnosis,
"severity": severity,
"color": color,
"avg_degradation": avg_degradation,
"stability_rate": stability_rate,
"test_score_std": test_score_std,
"recommendations": recommendations,
"summary": f"{color} {severity}: Degradation={avg_degradation:.2%}, "
f"Stability={stability_rate:.0%}"
}
ตัวอย่างการใช้งาน
if __name__ == "__main__":
from sklearn.ensemble import RandomForestRegressor
from sklearn.datasets import make_regression
# สร้างข้อมูลตัวอย่าง (มี noise และ pattern จริง)
np.random.seed(42)
X, y = make_regression(
n_samples=1000,
n_features=20,
noise=0.5,
random_state=42
)
# สร้างโมเดล generator
def model_gen():
return RandomForestRegressor(n_estimators=100, max_depth=5, random_state=42)
# รัน Analysis
diagnoser = OverfittingDiagnoser(n_splits=8, test_size=0.2)
results_df = diagnoser.run_walkforward_analysis(X, y, model_gen)
# แสดงรายงาน
report = diagnoser.get_diagnosis_report()
print("\n" + "=" * 50)
print("📊 DIAGNOSIS REPORT")
print("=" * 50)
print(report["summary"])
print("\n🔧 Recommendations:")
for rec in report["recommendations"]:
print(rec)เมตริกส์สำคัญในการวินิจฉัย Overfitting
| เมตริกส์ | คำอธิบาย | ค่าเฉพาะที่ดี | ค่าเฉลี่ยที่พบ |
|---|---|---|---|
| Degradation Rate | Train Score - Test Score | < 15% | 5-25% |
| Stability Rate | % ของ windows ที่ test > 0 | > 80% | 50-90% |
| Walk-forward Efficiency | Test Score / Train Score | > 0.85 | 0.60-0.95 |
| Consistency (CV) | Test Score Std / Mean | < 0.15 | 0.10-0.40 |
เทคนิคขั้นสูง: Nested Walk-forward
สำหรับระบบที่ซับซ้อน ควรใช้ Nested Walk-forward เพื่อหลีกเลี่ยงการ "peek" ข้อมูล
import itertools
from typing import List, Dict, Any
class NestedWalkForward:
"""
Nested Walk-forward Analysis
สำหรับ Parameter Optimization ที่ป้องกัน Overfitting
"""
def __init__(
self,
param_grid: Dict[str, List[Any]],
n_outer_splits: int = 5,
n_inner_splits: int = 3
):
self.param_grid = param_grid
self.n_outer_splits = n_outer_splits
self.n_inner_splits = n_inner_splits
self.best_params_history = []
def _generate_param_combinations(self) -> List[Dict]:
"""สร้าง combinations ของ parameters"""
keys = list(self.param_grid.keys())
values = list(self.param_grid.values())
combinations = list(itertools.product(*values))
return [dict(zip(keys, combo)) for combo in combinations]
def nested_cv_optimize(
self,
X: np.ndarray,
y: np.ndarray,
model_class,
scoring_func: callable
) -> Dict:
"""
Nested Cross-Validation สำหรับ Parameter Optimization
- Outer Loop: เลือก parameter ที่ดีที่สุด
- Inner Loop: ประเมิน parameter อย่าง fair
"""
from sklearn.model_selection import TimeSeriesSplit
tscv_outer = TimeSeriesSplit(n_splits=self.n_outer_splits)
all_outer_results = []
param_combinations = self._generate_param_combinations()
print(f"ทดสอบ {len(param_combinations)} parameter combinations")
for outer_idx, (train_val_idx, test_idx) in enumerate(tscv_outer.split(X)):
# แบ่ง train_val เป็น train และ validation สำหรับ inner CV
train_idx = train_val_idx[:int(len(train_val_idx) * 0.7)]
val_idx = train_val_idx[int(len(train_val_idx) * 0.7):]
X_train, X_val, X_test = X[train_idx], X[val_idx], X[test_idx]
y_train, y_val, y_test = y[train_idx], y[val_idx], y[test_idx]
best_score = -np.inf
best_params = None
# Inner CV: หา best parameters
for params in param_combinations:
model = model_class(**params)
model.fit(X_train, y_train)
val_score = scoring_func(model, X_val, y_val)
if val_score > best_score:
best_score = val_score
best_params = params.copy()
best_params['_inner_score'] = val_score
# Outer Test: ประเมิน best params บน test set ที่ไม่เคยเห็น
final_model = model_class(**best_params)
final_model.fit(X_train, y_train)
outer_test_score = scoring_func(final_model, X_test, y_test)
# ตรวจสอบ generalization gap
generalization_gap = best_params['_inner_score'] - outer_test_score
self.best_params_history.append({
'outer_fold': outer_idx + 1,
'best_params': {k: v for k, v in best_params.items() if k != '_inner_score'},
'inner_score': best_params['_inner_score'],
'outer_test_score': outer_test_score,
'generalization_gap': generalization_gap
})
print(f" Fold {outer_idx + 1}: Inner={best_params['_inner_score']:.4f}, "
f"Outer Test={outer_test_score:.4f}, Gap={generalization_gap:.4f}")
all_outer_results.append({
'params': best_params,
'test_score': outer_test_score,
'gap': generalization_gap
})
# สรุปผล
avg_test_score = np.mean([r['test_score'] for r in all_outer_results])
avg_gap = np.mean([r['gap'] for r in all_outer_results])
# เลือก params ที่ stable ที่สุด (ค่า gap ต่ำสุด)
most_stable = min(all_outer_results, key=lambda x: x['gap'])
return {
'average_test_score': avg_test_score,
'average_generalization_gap': avg_gap,
'most_stable_params': most_stable['params'],
'is_robust': avg_gap < 0.15 and avg_test_score > 0.5,
'history': self.best_params_history
}
ตัวอย่างการใช้งาน Nested Walk-forward
if __name__ == "__main__":
from sklearn.ensemble import GradientBoostingRegressor
# กำหนด parameter grid
param_grid = {
'n_estimators': [50, 100, 200],
'max_depth': [2, 3, 5],
'learning_rate': [0.01, 0.05, 0.1],
'min_samples_leaf': [5, 10, 20]
}
# สร้าง nested optimizer
optimizer = NestedWalkForward(
param_grid=param_grid,
n_outer_splits=5,
n_inner_splits=3
)
# รัน optimization
result = optimizer.nested_cv_optimize(
X=X, y=y,
model_class=GradientBoostingRegressor,
scoring_func=lambda m, X, y: m.score(X, y)
)
print("\n" + "=" * 50)
print("📊 NESTED WALK-FORWARD RESULTS")
print("=" * 50)
print(f"✅ Average Test Score: {result['average_test_score']:.4f}")
print(f"📉 Average Generalization Gap: {result['average_generalization_gap']:.4f}")
print(f"🎯 Robust Model: {'Yes' if result['is_robust'] else 'No'}")
print(f"⚙️ Most Stable Params: {result['most_stable_params']}")ใช้ AI วินิจฉัย Overfitting อัตโนมัติ
ปัจจุบันมี AI ที่ช่วยวิเคราะห์ผล Backtest และให้คำแนะนำแบบอัตโนมัติ HolySheep AI เป็นตัวเลือกที่น่าสนใจด้วยความเร็วต่ำกว่า 50ms และราคาประหยัดมาก
import requests
import json
from typing import Dict, List
import time
class HolySheepOverfittingAnalyzer:
"""
ใช้ HolySheep AI API วิเคราะห์ Overfitting อัตโนมัติ
"""
BASE_URL = "https://api.holysheep.ai/v1" # บังคับตามข้อกำหนด
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_backtest_results(
self,
walkforward_results: List[Dict],
model_description: str = ""
) -> Dict:
"""
วิเคราะห์ผล backtest ด้วย AI
Parameters:
-----------
walkforward_results : List[Dict]
ผลลัพธ์จาก Walk-forward Analysis
model_description : str
คำอธิบายโมเดลหรือระบบเทรด
"""
# คำนวณ statistics
train_scores = [r['train_score'] for r in walkforward_results]
test_scores = [r['test_score'] for r in walkforward_results]
stats = {
'avg_train_score': np.mean(train_scores),
'avg_test_score': np.mean(test_scores),
'std_test_score': np.std(test_scores),
'degradation_rate': np.mean([
r['train_score'] - r['test_score']
for r in walkforward_results
]),
'stability_rate': np.mean([
1 if r['test_score'] > 0 else 0
for r in walkforward_results
]),
'best_window': max(walkforward_results, key=lambda x: x['test_score']),
'worst_window': min(walkforward_results, key=lambda x: x['test_score'])
}
# สร้าง prompt สำหรับ AI
prompt = f"""คุณเป็นผู้เชี่ยวชาญด้าน Quantitative Trading และ Machine Learning
ข้อมูลโมเดล/ระบบ:
{model_description if model_description else 'ไม่ระบุ'}
ผลลัพธ์ Walk-forward Analysis:
{json.dumps(stats, indent=2)}
ผลลัพธ์ราย Window:
{json.dumps(walkforward_results, indent=2)}
กรุณาวิเคราะห์:
1. โมเดลนี้มี Overfitting หรือไม่ ระดับไหน?
2. สาเหตุที่เป็นไปได้คืออะไร?
3. แนะนำวิธีแก้ไขอย่างละเอียด
4. ควรใช้โมเดลนี้ในการเทรดจริงหรือไม่?
ตอบเป็นภาษาไทย พร้อมระดับความมั่นใจ (0-100%)"""
# เรียก HolySheep AI
start_time = time.time()
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json={
"model": "gpt-4.1", # หรือเลือกโมเดลอื่นตามความต้องการ
"messages": [
{"role": "system", "content": "คุณเป็นผู้เชี่ยวชาญด้าน Quantitative Trading"},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 2000
},
timeout=30
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
result = response.json()
ai_response = result['choices'][0]['message']['content']
usage = result.get('usage', {})
return {
'success': True,
'analysis': ai_response,
'latency_ms': round(latency_ms, 2),
'tokens_used': usage.get('total_tokens', 0),
'cost_usd': (usage.get('total_tokens', 0) / 1_000_000) * 8, # GPT-4.1: $8/MTok
'stats': stats
}
else:
return {
'success': False,
'error': response.text,
'latency_ms': round(latency_ms, 2)
}
def compare_models(
self,
models_results: Dict[str, List[Dict]]
) -> str:
"""
เปรียบเทียบหลายโมเดลพร้อมกัน
"""
prompt = "เปรียบเทียบโมเดลต่อไปนี้สำหรับการเทรดจริง:\n\n"
for model_name, results in models_results.items():
stats = {
'avg_test_score': np.mean([r['test_score'] for r in results]),
'degradation': np.mean([
r['train_score'] - r['test_score']
for r in results
]),
'stability': np.std([r['test_score'] for r in results])
}
prompt += f"## {model_name}:\n{json.dumps(stats, indent=2)}\n\n"
prompt += "จัดอันดับและแนะนำโมเดลที่เหมาะสมที่สุด"
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json={
"model": "claude-sonnet-4.5",
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.2
}
)
if response.status_code == 200:
return response.json()['choices'][0]['message']['content']
return f"Error: {response.status_code}"
ตัวอย่างการใช้งาน
if __name__ == "__main__":
# สมมติว่าได้ผลลัพธ์จาก Walk-forward Analysis แล้ว
sample_results = [
{'window': 1, 'train_score': 0.85, 'test_score': 0.72},
{'window': 2, 'train_score': 0.88, 'test_score': 0.68},
{'window': 3, 'train_score': 0.82, 'test_score': 0.75},
{'window': 4, 'train_score': 0.90, 'test_score': 0.65},
{'window': 5, 'train_score': 0.87, 'test_score': 0.71},
]
# เรียกใช้ Analyzer (ต้องใส่ API key จริง)
# analyzer = HolySheepOverfittingAnalyzer("YOUR_HOLYSHEEP_API_KEY")
# result = analyzer.analyze_backtest_results(
# walkforward_results=sample_results,
# model_description="LSTM Neural Network สำหรับ Predict ราคาหุ้น"
# )
# print(result['analysis'])
print("✅ HolySheep API Integration พร้อมใช้งาน")
print(f"📡 Base URL: https://api.holysheep.ai/v1")
print(f"⚡ Latency Target: <50ms")ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. Survivorship Bias — ข้อมูลมีเฉพาะบริษัทที่รอดชีวิต
อาการ: Backtest ดีมาก แต่ live ขาดทุนเพราะไม่ได้คิดถึงบริษัทที่ล้มเลิกกิจการไปแล้ว
# วิธีแก้ไข: ใช้ข้อมูลที่มี Survivorship Bias อยู่แล้ว หรือใช้ API ที่มีต่อมา
แนะนำ: ใช้ premium data provider ที่มี historical delisted stocks
ตัวอย่างการตรวจสอบ survivorship bias
def check_survivorship_bias(returns_df: pd.DataFrame, benchmark: pd.Series) -> Dict:
"""
ตรวจสอบว่าข้อมูลมี survivorship bias หรือไม่
"""
# คำนวณ systematic risk
systematic_return = benchmark.mean()
portfolio_return = returns_df.mean()
bias_ratio = (portfolio_return - systematic_return) / systematic_return
return {
'has_survivorship_bias': bias_ratio > 0.1, # ถ้าสูงกว่า 10% สงสัย