การทดสอบย้อนหลังการตัดสินใจของเอเจนต์

เหตุใดจึงต้องทดสอบย้อนหลังการตัดสินใจของเอเจนต์

กับดักอคติจากการใช้ข้อมูลอนาคต

import pandas as pd
import yfinance as yf

# WRONG — uses today's close to decide today's trade
def bad_signal(history: pd.DataFrame, date: str) -> str:
    today_close = history.loc[date, 'Close']  # look-ahead!
    if today_close > history['Close'].mean():
        return 'BUY'
    return 'HOLD'

# CORRECT — uses only data available BEFORE the decision date
def good_signal(history: pd.DataFrame, date: str) -> str:
    past_data = history.loc[:date].iloc[:-1]  # exclude today
    if past_data.empty:
        return 'HOLD'
    today_open = history.loc[date, 'Open']    # open price known at market open
    if today_open > past_data['Close'].mean():
        return 'BUY'
    return 'HOLD'

สร้างลูปเล่นซ้ำข้อมูลย้อนหลัง

import yfinance as yf
import pandas as pd

def backtest_simple(ticker: str, start: str, end: str) -> list[dict]:
    hist = yf.Ticker(ticker).history(start=start, end=end)
    results = []

    for i in range(20, len(hist)):  # need 20 days of history for signals
        window   = hist.iloc[:i]           # data available on day i
        today    = hist.iloc[i]
        decision = good_signal(window, str(hist.index[i].date()))

        results.append({
            'date':     str(hist.index[i].date()),
            'decision': decision,
            'open':     float(today['Open']),
            'close':    float(today['Close']),
            'next_day_return': None  # filled in next iteration
        })
    return results

จำลองตรรกะของเอเจนต์ในแต่ละขั้นตอน

import openai, json
import yfinance as yf

client = openai.OpenAI(api_key='YOUR_OPENAI_KEY')

def agent_decision_at_date(ticker: str, hist_up_to: pd.DataFrame) -> str:
    recent = hist_up_to.tail(20)
    closes = recent['Close'].tolist()
    returns = [round((closes[i] - closes[i-1]) / closes[i-1], 4)
               for i in range(1, len(closes))]
    prompt = (
        f'Ticker: {ticker}\n'
        f'Last 20 closing prices: {closes}\n'
        f'Daily returns: {returns}\n'
        f'Based on this data, recommend: BUY, SELL, or HOLD.\n'
        f'Return JSON: {{"action": "BUY/SELL/HOLD", "reason": "..."}}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    return json.loads(resp.choices[0].message.content).get('action', 'HOLD')

คำนวณอัตราการทายถูก

def compute_hit_rate(results: list[dict]) -> dict:
    buy_decisions = [r for r in results if r['decision'] == 'BUY'
                     and r.get('next_day_return') is not None]

    if not buy_decisions:
        return {'hit_rate': None, 'sample_size': 0}

    correct = sum(1 for r in buy_decisions if r['next_day_return'] > 0)
    return {
        'hit_rate':     round(correct / len(buy_decisions), 4),
        'sample_size':  len(buy_decisions),
        'correct':      correct,
        'interpretation': 'above_chance' if correct / len(buy_decisions) > 0.52 else 'at_or_below_chance'
    }

if __name__ == '__main__':
    results = [
        {'decision': 'BUY', 'next_day_return': 0.012},
        {'decision': 'BUY', 'next_day_return': -0.004},
        {'decision': 'HOLD', 'next_day_return': None},
        {'decision': 'BUY', 'next_day_return': 0.007},
    ]
    print('Hit rate:', compute_hit_rate(results))

คำนวณผลตอบแทนเทียบเกณฑ์มาตรฐาน

def compare_to_benchmark(results: list[dict], ticker: str, period: str = '1y') -> dict:
    import yfinance as yf

    # Strategy return: only invested on BUY days
    buy_days = [r for r in results if r['decision'] == 'BUY'
                and r.get('next_day_return') is not None]
    strategy_return = sum(r['next_day_return'] for r in buy_days)

    # Benchmark: buy-and-hold
    hist = yf.Ticker(ticker).history(period=period)['Close']
    bh_return = float((hist.iloc[-1] - hist.iloc[0]) / hist.iloc[0])

    return {
        'strategy_return_pct':  round(strategy_return * 100, 2),
        'benchmark_return_pct': round(bh_return * 100, 2),
        'outperformed':         strategy_return > bh_return,
        'alpha_pct':            round((strategy_return - bh_return) * 100, 2)
    }

ผลกระทบของต้นทุนธุรกรรม

COMMISSION_PCT = 0.001   # 0.1% per trade (buy + sell)
SLIPPAGE_PCT   = 0.0005  # 0.05% per trade

def returns_after_costs(decisions: list[dict]) -> dict:
    total_return = 0.0
    total_cost   = 0.0

    for d in decisions:
        if d['decision'] == 'BUY' and d.get('next_day_return') is not None:
            gross = d['next_day_return']
            cost  = COMMISSION_PCT + SLIPPAGE_PCT
            net   = gross - cost
            total_return += net
            total_cost   += cost

    return {
        'gross_return_pct': round((total_return + total_cost) * 100, 2),
        'net_return_pct':   round(total_return * 100, 2),
        'total_costs_pct':  round(total_cost * 100, 2),
        'num_trades':       sum(1 for d in decisions if d['decision'] == 'BUY')
    }

if __name__ == '__main__':
    decisions = [
        {'decision': 'BUY', 'next_day_return': 0.012},
        {'decision': 'BUY', 'next_day_return': -0.004},
        {'decision': 'HOLD', 'next_day_return': None},
    ]
    print('Returns after trading costs:', returns_after_costs(decisions))

การตรวจสอบแบบเดินหน้า

def walk_forward_test(ticker: str, total_years: int = 3,
                      train_months: int = 12, test_months: int = 3) -> list[dict]:
    import yfinance as yf
    from dateutil.relativedelta import relativedelta
    from datetime import date

    results = []
    start = date(date.today().year - total_years, 1, 1)
    end   = date.today()

    window_start = start
    while True:
        train_end = window_start + relativedelta(months=train_months)
        test_end  = train_end + relativedelta(months=test_months)
        if test_end > end:
            break

        # In practice: train your signal on window_start->train_end
        # then evaluate on train_end->test_end
        results.append({
            'train_period': f'{window_start} to {train_end}',
            'test_period':  f'{train_end} to {test_end}'
        })
        window_start += relativedelta(months=test_months)

    return results

รายงานการทดสอบย้อนหลัง

def backtest_report(ticker: str, results: list[dict]) -> dict:
    hit = compute_hit_rate(results)
    benchmark = compare_to_benchmark(results, ticker)
    costs = returns_after_costs(results)

    return {
        'ticker':                  ticker,
        'total_decisions':         len(results),
        'buy_signals':             costs['num_trades'],
        'hit_rate':                hit.get('hit_rate'),
        'strategy_return_gross':   f'{costs["gross_return_pct"]}%',
        'strategy_return_net':     f'{costs["net_return_pct"]}%',
        'benchmark_return':        f'{benchmark["benchmark_return_pct"]}%',
        'alpha':                   f'{benchmark["alpha_pct"]}%',
        'outperformed_benchmark':  benchmark['outperformed'],
        'transaction_costs':       f'{costs["total_costs_pct"]}%'
    }

ตีความผลการทดสอบย้อนหลัง

def interpret_backtest(report: dict) -> list[str]:
    warnings = []

    hit_rate = report.get('hit_rate')
    if hit_rate and hit_rate < 0.50:
        warnings.append(f'Hit rate {hit_rate:.1%} is below chance — strategy predicts direction poorly')

    alpha_str = report.get('alpha', '0%')
    alpha = float(alpha_str.replace('%', '')) / 100
    if alpha < 0:
        warnings.append(f'Negative alpha {alpha_str}: strategy underperforms buy-and-hold')

    if not warnings:
        return ['Backtest results look acceptable — proceed with paper trading before live deployment']
    return warnings

if __name__ == '__main__':
    report = {'hit_rate': 0.45, 'alpha': '-1.2%'}
    for warning in interpret_backtest(report):
        print('-', warning)

การจำลองมอนติคาร์โลเพื่อทดสอบความทนทาน

import numpy as np

def monte_carlo_backtest(daily_returns: list[float],
                         simulations: int = 1000) -> dict:
    returns_arr = np.array(daily_returns)
    final_returns = []

    for _ in range(simulations):
        shuffled = np.random.choice(returns_arr, size=len(returns_arr), replace=True)
        cumulative = float((1 + shuffled).prod() - 1)
        final_returns.append(cumulative)

    final_returns = sorted(final_returns)
    return {
        'median_return':      round(float(np.median(final_returns)) * 100, 2),
        'percentile_5':       round(float(np.percentile(final_returns, 5)) * 100, 2),
        'percentile_95':      round(float(np.percentile(final_returns, 95)) * 100, 2),
        'prob_positive':      round(sum(1 for r in final_returns if r > 0) / simulations, 3)
    }

อคติจากการใช้ข้อมูลอนาคตในการทดสอบย้อนหลังคืออะไร

สรุปการทดสอบย้อนหลังการตัดสินใจของเอเจนต์