วิธีตั้ง trailing stop streaming

Trailing Stop

Trailing Stop คำสั่ง Stop Loss เคลื่อนตามราคาอัตโนมัติ ราคาขึ้น Stop ขยับขึ้นตาม ราคาลง Stop อยู่ที่เดิม ล็อกกำไรอัตโนมัติ ตั้งเป็น % หรือจำนวนเงิน

Streaming Data รับข้อมูล Real-time WebSocket SSE ราคาหุ้นคริปโต Forex เปลี่ยนทุกวินาที ลด Latency Binance API TradingView Alpaca

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 อยู่ที่เดิม ล็อกกำไรอัตโนมัติ ตั้งเป็น % หรือจำนวนเงิน

สรุป

Trailing Stop เคลื่อนตามราคาล็อกกำไรอัตโนมัติ WebSocket Streaming Real-time Binance Alpaca ATR คำนวณ Stop แม่นยำ Position Sizing Risk Management 2% Rule Backtest ก่อนใช้จริง