วิธีตั้ง Trailing Stop กับ Streaming —

Trailing Stop

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

เนื้อหาเกี่ยวข้อง — อ่านต่อ: Directus CMS Capacity Planning — คู่มือฉบับสมบูรณ์ 2026

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

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน Distributed Tracing Low Code No Code

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

แนะนำเพิ่มเติม — คู่มือเทรดจาก SiamCafeBook

เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน Libvirt KVM Chaos Engineering — คู่มือฉบับสมบูรณ์ 2026