trend following books

Trend Following Strategy

Trend Following เป็น Trading Strategy ที่ตามทิศทางของตลาด ซื้อเมื่อขาขึ้น ขายเมื่อขาลง ใช้ Technical Indicators เช่น Moving Average, Breakout ไม่พยากรณ์ตลาด ทำกำไรจาก Big Trends

หนังสือเกี่ยวกับ Trend Following ช่วยเรียนรู้แนวคิด Trading Systems Risk Management และวิธี Backtest Strategy ด้วยข้อมูลจริง

Trend Following Backtesting ด้วย Python

# trend_following.py — Trend Following Backtesting
# pip install yfinance pandas numpy matplotlib

import numpy as np
import pandas as pd
import yfinance as yf
from datetime import datetime

class TrendFollowingStrategy:
    """Trend Following Strategy ด้วย Moving Average Crossover"""

    def __init__(self, fast_period=50, slow_period=200, atr_period=20, risk_pct=0.02):
        self.fast_period = fast_period
        self.slow_period = slow_period
        self.atr_period = atr_period
        self.risk_pct = risk_pct  # Risk 2% per trade

    def calculate_signals(self, df):
        """คำนวณ Trading Signals"""
        df = df.copy()

        # Moving Averages
        df['SMA_fast'] = df['Close'].rolling(self.fast_period).mean()
        df['SMA_slow'] = df['Close'].rolling(self.slow_period).mean()

        # ATR (Average True Range) สำหรับ Position Sizing
        high_low = df['High'] - df['Low']
        high_close = (df['High'] - df['Close'].shift()).abs()
        low_close = (df['Low'] - df['Close'].shift()).abs()
        tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
        df['ATR'] = tr.rolling(self.atr_period).mean()

        # Signals: 1 = Long, -1 = Short, 0 = No position
        df['Signal'] = 0
        df.loc[df['SMA_fast'] > df['SMA_slow'], 'Signal'] = 1
        df.loc[df['SMA_fast'] < df['SMA_slow'], 'Signal'] = -1

        # Position Changes
        df['Position'] = df['Signal'].diff().fillna(0)

        return df

    def backtest(self, df, initial_capital=100000):
        """Backtest Strategy"""
        df = self.calculate_signals(df)
        df = df.dropna()

        capital = initial_capital
        position = 0
        shares = 0
        trades = []
        equity_curve = []

        for i, row in df.iterrows():
            # Position Sizing ด้วย ATR
            if row['ATR'] > 0:
                position_size = (capital * self.risk_pct) / row['ATR']
            else:
                position_size = 0

            # Entry
            if row['Position'] != 0 and position == 0:
                shares = int(position_size)
                if shares > 0:
                    entry_price = row['Close']
                    position = row['Signal']
                    cost = shares * entry_price
                    trades.append({
                        'entry_date': i, 'entry_price': entry_price,
                        'shares': shares, 'direction': 'Long' if position > 0 else 'Short',
                    })

            # Exit
            elif row['Position'] != 0 and position != 0:
                exit_price = row['Close']
                if position > 0:
                    pnl = (exit_price - entry_price) * shares
                else:
                    pnl = (entry_price - exit_price) * shares
                capital += pnl
                if trades:
                    trades[-1].update({
                        'exit_date': i, 'exit_price': exit_price, 'pnl': pnl,
                    })
                position = 0
                shares = 0

            # Equity
            if position != 0:
                if position > 0:
                    unrealized = (row['Close'] - entry_price) * shares
                else:
                    unrealized = (entry_price - row['Close']) * shares
                equity_curve.append(capital + unrealized)
            else:
                equity_curve.append(capital)

        return pd.DataFrame(trades), pd.Series(equity_curve, index=df.index[:len(equity_curve)])

    def performance_report(self, trades_df, equity):
        """สร้าง Performance Report"""
        if trades_df.empty:
            print("No trades")
            return

        completed = trades_df.dropna(subset=['pnl'])

        total_trades = len(completed)
        winners = completed[completed['pnl'] > 0]
        losers = completed[completed['pnl'] <= 0]
        win_rate = len(winners) / total_trades * 100 if total_trades > 0 else 0

        total_pnl = completed['pnl'].sum()
        avg_win = winners['pnl'].mean() if len(winners) > 0 else 0
        avg_loss = losers['pnl'].mean() if len(losers) > 0 else 0

        # Sharpe Ratio
        returns = equity.pct_change().dropna()
        sharpe = returns.mean() / returns.std() * np.sqrt(252) if returns.std() > 0 else 0

        # Max Drawdown
        peak = equity.expanding().max()
        drawdown = (equity - peak) / peak
        max_dd = drawdown.min() * 100

        print(f"\n{'='*55}")
        print(f"Trend Following Performance Report")
        print(f"{'='*55}")
        print(f"  Total Trades: {total_trades}")
        print(f"  Win Rate: {win_rate:.1f}%")
        print(f"  Total P&L: ")
        print(f"  Avg Win: ")
        print(f"  Avg Loss: ")
        print(f"  Profit Factor: {abs(avg_win/avg_loss):.2f}" if avg_loss != 0 else "")
        print(f"  Sharpe Ratio: {sharpe:.2f}")
        print(f"  Max Drawdown: {max_dd:.1f}%")
        print(f"  Final Equity: ")

# ตัวอย่าง
# data = yf.download("SPY", start="2010-01-01", end="2024-01-01")
# strategy = TrendFollowingStrategy(fast_period=50, slow_period=200)
# trades, equity = strategy.backtest(data, initial_capital=100000)
# strategy.performance_report(trades, equity)

print("Trend Following Backtester")
print("  Strategy: SMA 50/200 Crossover")
print("  Risk: 2% per trade (ATR-based)")
print("  Markets: SPY, QQQ, Futures")

Breakout Strategy

# breakout_strategy.py — Donchian Channel Breakout (Turtle Trading)
# pip install pandas numpy

import numpy as np
import pandas as pd

class DonchianBreakout:
    """Donchian Channel Breakout Strategy (Turtle Trading)"""

    def __init__(self, entry_period=20, exit_period=10, atr_period=20, risk_pct=0.01):
        self.entry_period = entry_period
        self.exit_period = exit_period
        self.atr_period = atr_period
        self.risk_pct = risk_pct

    def calculate_channels(self, df):
        """คำนวณ Donchian Channels"""
        df = df.copy()

        # Entry Channel (20-day)
        df['Entry_High'] = df['High'].rolling(self.entry_period).max()
        df['Entry_Low'] = df['Low'].rolling(self.entry_period).min()

        # Exit Channel (10-day)
        df['Exit_High'] = df['High'].rolling(self.exit_period).max()
        df['Exit_Low'] = df['Low'].rolling(self.exit_period).min()

        # ATR
        high_low = df['High'] - df['Low']
        high_close = (df['High'] - df['Close'].shift()).abs()
        low_close = (df['Low'] - df['Close'].shift()).abs()
        tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
        df['ATR'] = tr.rolling(self.atr_period).mean()

        return df

    def generate_signals(self, df):
        """สร้าง Trading Signals"""
        df = self.calculate_channels(df)
        df = df.dropna()

        signals = []
        position = 0

        for i in range(1, len(df)):
            row = df.iloc[i]
            prev = df.iloc[i - 1]

            # Long Entry: Close ทะลุ Entry_High
            if position == 0 and row['Close'] > prev['Entry_High']:
                position = 1
                signals.append({
                    'date': df.index[i],
                    'action': 'BUY',
                    'price': row['Close'],
                    'atr': row['ATR'],
                })

            # Long Exit: Close ต่ำกว่า Exit_Low
            elif position == 1 and row['Close'] < prev['Exit_Low']:
                position = 0
                signals.append({
                    'date': df.index[i],
                    'action': 'SELL',
                    'price': row['Close'],
                    'atr': row['ATR'],
                })

            # Short Entry: Close ทะลุ Entry_Low
            elif position == 0 and row['Close'] < prev['Entry_Low']:
                position = -1
                signals.append({
                    'date': df.index[i],
                    'action': 'SHORT',
                    'price': row['Close'],
                    'atr': row['ATR'],
                })

            # Short Exit: Close สูงกว่า Exit_High
            elif position == -1 and row['Close'] > prev['Exit_High']:
                position = 0
                signals.append({
                    'date': df.index[i],
                    'action': 'COVER',
                    'price': row['Close'],
                    'atr': row['ATR'],
                })

        return pd.DataFrame(signals)

# Turtle Trading Rules Summary
turtle_rules = {
    "Entry": "Buy when price breaks 20-day high, Short when breaks 20-day low",
    "Exit": "Exit long below 10-day low, Exit short above 10-day high",
    "Position Size": "1 ATR = 1% of account (N = ATR(20))",
    "Adding": "Add every 0.5 ATR up to 4 units",
    "Stop Loss": "2 ATR from entry price",
    "Markets": "Diversified: Bonds, Currencies, Commodities, Indices",
    "Risk": "Max 2% per position, 6% per correlated group",
}

print("Turtle Trading Rules:")
for key, value in turtle_rules.items():
    print(f"  {key}: {value}")

Portfolio Risk Management

# risk_management.py — Portfolio Risk Management
import numpy as np
from dataclasses import dataclass
from typing import List

@dataclass
class Position:
    symbol: str
    direction: str  # long, short
    entry_price: float
    current_price: float
    shares: int
    atr: float
    sector: str

class RiskManager:
    """Portfolio Risk Management สำหรับ Trend Following"""

    def __init__(self, total_capital, max_risk_per_trade=0.02,
                 max_risk_per_sector=0.06, max_total_risk=0.20):
        self.capital = total_capital
        self.max_risk_per_trade = max_risk_per_trade
        self.max_risk_per_sector = max_risk_per_sector
        self.max_total_risk = max_total_risk
        self.positions: List[Position] = []

    def calculate_position_size(self, price, atr):
        """Position Size ตาม ATR"""
        risk_amount = self.capital * self.max_risk_per_trade
        if atr <= 0:
            return 0
        shares = int(risk_amount / atr)
        max_cost = self.capital * 0.10  # Max 10% ต่อ Position
        max_shares = int(max_cost / price)
        return min(shares, max_shares)

    def can_add_position(self, sector):
        """ตรวจสอบว่าเพิ่ม Position ได้หรือไม่"""
        # Sector Risk
        sector_risk = sum(
            abs(p.shares * p.atr) for p in self.positions
            if p.sector == sector
        ) / self.capital
        if sector_risk >= self.max_risk_per_sector:
            return False, f"Sector risk limit ({sector}): {sector_risk:.1%}"

        # Total Risk
        total_risk = sum(
            abs(p.shares * p.atr) for p in self.positions
        ) / self.capital
        if total_risk >= self.max_total_risk:
            return False, f"Total risk limit: {total_risk:.1%}"

        return True, "OK"

    def portfolio_report(self):
        """Portfolio Report"""
        print(f"\n{'='*55}")
        print(f"Portfolio Risk Report")
        print(f"{'='*55}")
        print(f"  Capital: ")
        print(f"  Positions: {len(self.positions)}")

        total_exposure = 0
        total_pnl = 0

        for p in self.positions:
            if p.direction == "long":
                pnl = (p.current_price - p.entry_price) * p.shares
            else:
                pnl = (p.entry_price - p.current_price) * p.shares
            exposure = p.shares * p.current_price
            total_exposure += exposure
            total_pnl += pnl
            risk = p.shares * p.atr * 2  # 2 ATR stop

            print(f"  {p.symbol:>6} {p.direction:>5} | "
                  f"Entry:  | "
                  f"P&L:  | "
                  f"Risk: ")

        print(f"\n  Total Exposure:  "
              f"({total_exposure/self.capital:.0%})")
        print(f"  Total P&L: ")

# ตัวอย่าง
rm = RiskManager(total_capital=1000000)

positions = [
    Position("SPY", "long", 450, 465, 200, 5.2, "Equity"),
    Position("GLD", "long", 180, 195, 500, 2.8, "Commodity"),
    Position("TLT", "short", 100, 92, 800, 1.5, "Bond"),
    Position("EURUSD", "long", 1.08, 1.10, 50000, 0.008, "Currency"),
]

for p in positions:
    rm.positions.append(p)

rm.portfolio_report()

# Position Sizing
atr = 5.0
size = rm.calculate_position_size(price=450, atr=atr)
print(f"\nNew Position Size: {size} shares (ATR={atr})")

หนังสือแนะนำ

• Trend Following (Covel): แนวคิดพื้นฐาน ประวัติ Trend Followers ที่สำเร็จ เหมาะเริ่มต้น
• Following the Trend (Clenow): Managed Futures Diversified Portfolio พร้อม Code
• Way of the Turtle (Faith): Turtle Trading Rules จากประสบการณ์ตรง
• Systematic Trading (Carver): Portfolio Construction Risk Management สำหรับ Systematic Traders
• Trading Systems and Methods (Kaufman): Technical Reference ครอบคลุมทุก Trading System
• Stocks on the Move (Clenow): Momentum Strategy สำหรับ Stocks พร้อม Backtesting

Trend Following คืออะไร

Trading Strategy ซื้อขาขึ้น ขายขาลง ใช้ Moving Average Breakout Momentum ไม่พยากรณ์ตลาด ตามทิศทางที่เกิดขึ้นแล้ว ทำกำไรจาก Big Trends

สรุป

Trend Following เป็น Strategy ที่พิสูจน์แล้วว่าทำกำไรได้ระยะยาว เรียนรู้จากหนังสือ Michael Covel Andreas Clenow Curtis Faith Robert Carver ใช้ Python Backtest Strategy SMA Crossover Donchian Breakout ATR Position Sizing Risk Management กระจาย Portfolio หลายตลาด