วิธีตั้ง Trailing Stop กับ Streaming —
Trailing Stop
Trailing Stop คำสั่ง Stop Loss เคลื่อนตามราคาอัตโนมัติ ราคาขึ้น Stop ขยับขึ้นตาม ราคาลง Stop อยู่ที่เดิม ล็อกกำไรอัตโนมัติ ตั้งเป็น % หรือจำนวนเงิน
Streaming Data รับข้อมูล Real-time WebSocket SSE ราคาหุ้นคริปโต Forex เปลี่ยนทุกวินาที ลด Latency Binance API TradingView Alpaca
| สไตล์ | Trailing % | ระยะเวลา | เหมาะกับ |
|---|---|---|---|
| Scalping | 0.5-1% | นาที-ชั่วโมง | Forex, Crypto |
| Day Trading | 1-3% | ชั่วโมง-วัน | หุ้น, Crypto |
| Swing Trading | 5-10% | วัน-สัปดาห์ | หุ้น, ETF |
| Position | 10-20% | สัปดาห์-เดือน | หุ้น, กองทุน |
Trailing Stop Bot
# trailing_stop.py — Trailing Stop Bot with Streaming
# pip install websockets aiohttp
import asyncio
from dataclasses import dataclass, field
from typing import Optional, List, Dict
from datetime import datetime
@dataclass
class Position:
symbol: str
entry_price: float
quantity: float
trailing_pct: float
highest_price: float = 0
stop_price: float = 0
status: str = "open" # open, stopped, manual_close
pnl: float = 0
def __post_init__(self):
self.highest_price = self.entry_price
self.stop_price = self.entry_price * (1 - self.trailing_pct / 100)
def update(self, current_price: float) -> dict:
"""อัปเดต Trailing Stop ตามราคาปัจจุบัน"""
result = {"action": "hold", "price": current_price}
if current_price > self.highest_price:
# ราคาขึ้น: ขยับ Stop ตาม
self.highest_price = current_price
self.stop_price = current_price * (1 - self.trailing_pct / 100)
result["action"] = "trail_up"
result["new_stop"] = self.stop_price
elif current_price <= self.stop_price:
# ราคาลงถึง Stop: ขาย
self.status = "stopped"
self.pnl = (current_price - self.entry_price) * self.quantity
result["action"] = "stop_triggered"
result["pnl"] = self.pnl
return result
class TrailingStopBot:
"""Trailing Stop Bot with Real-time Streaming"""
def __init__(self):
self.positions: Dict[str, Position] = {}
self.trade_history: List[dict] = []
def open_position(self, symbol: str, price: float,
quantity: float, trailing_pct: float):
"""เปิด Position ใหม่"""
pos = Position(symbol, price, quantity, trailing_pct)
self.positions[symbol] = pos
print(f" OPEN: {symbol} @ {price:.2f} x {quantity}")
print(f" Trailing: {trailing_pct}% | Stop: {pos.stop_price:.2f}")
def process_tick(self, symbol: str, price: float) -> Optional[dict]:
"""ประมวลผลราคาใหม่"""
pos = self.positions.get(symbol)
if not pos or pos.status != "open":
return None
result = pos.update(price)
if result["action"] == "trail_up":
print(f" TRAIL: {symbol} High={pos.highest_price:.2f} "
f"Stop={pos.stop_price:.2f}")
elif result["action"] == "stop_triggered":
pnl_pct = (price - pos.entry_price) / pos.entry_price * 100
print(f" STOP: {symbol} @ {price:.2f} "
f"PnL: {pos.pnl:+.2f} ({pnl_pct:+.1f}%)")
self.trade_history.append({
"symbol": symbol,
"entry": pos.entry_price,
"exit": price,
"pnl": pos.pnl,
"pnl_pct": pnl_pct,
})
return result
def show_positions(self):
print(f"\n{'='*55}")
print(f"Open Positions")
print(f"{'='*55}")
for symbol, pos in self.positions.items():
if pos.status == "open":
print(f" {symbol}: Entry={pos.entry_price:.2f} "
f"High={pos.highest_price:.2f} "
f"Stop={pos.stop_price:.2f} "
f"Trail={pos.trailing_pct}%")
# ตัวอย่าง
bot = TrailingStopBot()
bot.open_position("BTC/USDT", 67500, 0.1, 5)
bot.open_position("ETH/USDT", 3200, 1.0, 7)
# จำลอง Price Stream
btc_prices = [67500, 68000, 69500, 71000, 72500, 73000, 72000, 70000, 69500]
print(f"\nBTC/USDT Price Stream:")
for price in btc_prices:
result = bot.process_tick("BTC/USDT", price)
if result and result["action"] == "stop_triggered":
break
bot.show_positions()WebSocket Streaming
# websocket_stream.py — WebSocket Price Streaming
import asyncio
import json
# Binance WebSocket Stream
# async def binance_stream(symbol: str, bot: TrailingStopBot):
# """เชื่อมต่อ Binance WebSocket รับราคา Real-time"""
# import websockets
# url = f"wss://stream.binance.com:9443/ws/{symbol.lower()}@trade"
#
# async with websockets.connect(url) as ws:
# print(f" Connected to Binance {symbol} stream")
# async for message in ws:
# data = json.loads(message)
# price = float(data["p"])
# result = bot.process_tick(symbol.upper(), price)
# if result and result["action"] == "stop_triggered":
# print(f" Stop triggered! Closing connection...")
# break
# Alpaca WebSocket Stream
# async def alpaca_stream(symbol: str, bot: TrailingStopBot):
# """เชื่อมต่อ Alpaca WebSocket"""
# import websockets
# url = "wss://stream.data.alpaca.markets/v2/iex"
#
# async with websockets.connect(url) as ws:
# # Authenticate
# auth = {"action": "auth", "key": "API_KEY", "secret": "API_SECRET"}
# await ws.send(json.dumps(auth))
#
# # Subscribe
# sub = {"action": "subscribe", "trades": [symbol]}
# await ws.send(json.dumps(sub))
#
# async for message in ws:
# data = json.loads(message)
# for trade in data:
# if trade.get("T") == "t":
# price = trade["p"]
# bot.process_tick(symbol, price)
# ATR-based Trailing Stop
class ATRTrailingStop:
"""Trailing Stop based on ATR (Average True Range)"""
def __init__(self, atr_period: int = 14, atr_multiplier: float = 2.0):
self.atr_period = atr_period
self.atr_multiplier = atr_multiplier
self.highs: list = []
self.lows: list = []
self.closes: list = []
def add_candle(self, high: float, low: float, close: float):
"""เพิ่มข้อมูลแท่งเทียน"""
self.highs.append(high)
self.lows.append(low)
self.closes.append(close)
def calculate_atr(self) -> float:
"""คำนวณ ATR"""
if len(self.closes) < self.atr_period + 1:
return 0
true_ranges = []
for i in range(-self.atr_period, 0):
tr = max(
self.highs[i] - self.lows[i],
abs(self.highs[i] - self.closes[i-1]),
abs(self.lows[i] - self.closes[i-1]),
)
true_ranges.append(tr)
return sum(true_ranges) / len(true_ranges)
def get_stop_price(self, current_price: float) -> float:
"""คำนวณ Stop Price จาก ATR"""
atr = self.calculate_atr()
if atr == 0:
return current_price * 0.95 # Default 5%
return current_price - (atr * self.atr_multiplier)
# ตัวอย่าง ATR
atr_stop = ATRTrailingStop(atr_period=14, atr_multiplier=2.0)
# จำลอง 20 แท่งเทียน
import random
random.seed(42)
price = 100
for i in range(20):
change = random.uniform(-3, 3)
high = price + abs(change)
low = price - abs(change)
close = price + change
atr_stop.add_candle(high, low, close)
price = close
atr = atr_stop.calculate_atr()
stop = atr_stop.get_stop_price(price)
print(f"\nATR Trailing Stop:")
print(f" Current Price: {price:.2f}")
print(f" ATR(14): {atr:.2f}")
print(f" ATR Stop (2x): {stop:.2f}")
print(f" Distance: {price - stop:.2f} ({(price - stop)/price*100:.1f}%)")
# Streaming Comparison
streaming = {
"WebSocket": {"latency": "< 10ms", "use": "Binance, Alpaca, TradingView"},
"SSE": {"latency": "< 100ms", "use": "Custom API, Dashboard"},
"REST Polling": {"latency": "1-5s", "use": "Legacy API, Backup"},
"gRPC Stream": {"latency": "< 5ms", "use": "Internal Services"},
}
print(f"\nStreaming Methods:")
for method, info in streaming.items():
print(f" {method}: Latency {info['latency']} | Use: {info['use']}")Risk Management
# risk_management.py — Risk Management for Trading
from dataclasses import dataclass
@dataclass
class RiskParams:
account_balance: float
risk_per_trade_pct: float # % ของพอร์ตที่ยอมเสีย
trailing_stop_pct: float
@property
def risk_amount(self) -> float:
return self.account_balance * self.risk_per_trade_pct / 100
def position_size(self, entry_price: float) -> float:
"""คำนวณขนาด Position ตาม Risk"""
stop_distance = entry_price * self.trailing_stop_pct / 100
if stop_distance == 0:
return 0
return self.risk_amount / stop_distance
def show(self, entry_price: float):
size = self.position_size(entry_price)
stop = entry_price * (1 - self.trailing_stop_pct / 100)
value = size * entry_price
print(f"\n Risk Management:")
print(f" Account: {self.account_balance:,.0f} USDT")
print(f" Risk/Trade: {self.risk_per_trade_pct}% = {self.risk_amount:,.0f} USDT")
print(f" Entry: {entry_price:,.2f}")
print(f" Trailing Stop: {self.trailing_stop_pct}%")
print(f" Stop Price: {stop:,.2f}")
print(f" Position Size: {size:.4f}")
print(f" Position Value: {value:,.2f} USDT")
# ตัวอย่าง
risk = RiskParams(
account_balance=10000,
risk_per_trade_pct=2, # ยอมเสีย 2% ต่อ Trade
trailing_stop_pct=5, # Trailing Stop 5%
)
risk.show(entry_price=67500) # BTC
# Risk Rules
rules = {
"1% Rule": "ยอมเสียไม่เกิน 1% ของพอร์ตต่อ Trade (Conservative)",
"2% Rule": "ยอมเสียไม่เกิน 2% ของพอร์ตต่อ Trade (Standard)",
"6% Rule": "ยอมเสียรวมไม่เกิน 6% ของพอร์ตต่อเดือน",
"Risk:Reward": "อย่างน้อย 1:2 เช่น เสี่ยง 100 ต้องมีโอกาสได้ 200",
"Max Positions": "ไม่เกิน 5 Positions พร้อมกัน",
"Correlation": "ไม่ถือ Positions ที่ Correlated กันเกิน 3",
}
print(f"\n\nRisk Rules:")
for rule, desc in rules.items():
print(f" {rule}: {desc}")เคล็ดลับ
- ATR Stop: ใช้ ATR คำนวณ Trailing Stop แม่นยำกว่า % คงที่
- WebSocket: ใช้ WebSocket รับราคา Real-time Latency ต่ำ
- Position Size: คำนวณขนาดจาก Risk ไม่ใช่จากเงินที่มี
- 2% Rule: ยอมเสียไม่เกิน 2% ต่อ Trade ป้องกัน Drawdown
- Backtest: ทดสอบ Trailing Stop % กับข้อมูลย้อนหลังก่อนใช้จริง
- ไม่ขยับ Stop ลง: Trailing Stop ต้องขึ้นอย่างเดียว ห้ามขยับลง
Trailing Stop คืออะไร
Stop Loss เคลื่อนตามราคาอัตโนมัติ ราคาขึ้น Stop ขยับขึ้นตาม ราคาลง Stop อยู่ที่เดิม ล็อกกำไรอัตโนมัติ ตั้งเป็น % หรือจำนวนเงิน