Trend Following Books — หนังสือแนะนำเกี่ยวกับ

Trend Following Strategy

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

เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง MQL5 — คู่มือเทรด Forex ฉบับสมบูรณ์ 2026

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

เนื้อหาเกี่ยวข้อง — อ่านต่อ: yen carry trade crash

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

แนะนำเพิ่มเติม — หนังสือเทรดที่ SiamCafeBook

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน Tcci MQL5 — คู่มือเทรด Forex ฉบับสมบูรณ์ 2026