| Logic Chuyển Trạng Thái Chi Tiết FSM

Bài viết này — bản mở rộng đầy đủ — đi sâu vào logic chuyển trạng thái chi tiết của FSM trong bot trading: bảng chuyển trạng thái, điều kiện, xử lý tình huống biên, code mẫu và kiểm thử.

1. Bảng chuyển trạng thái (State Transition Table)

Thiết kế FSM chuyên nghiệp bắt đầu từ bảng: mỗi dòng là “từ trạng thái + sự kiện + điều kiện → trạng thái mới + hành động”.

Trạng thái hiện tại	Sự kiện	Điều kiện	Trạng thái mới	Hành động
IDLE	tín hiệu BUY	spread ok, ADX > 25	ENTER	kiểm tra bộ lọc
ENTER	đủ điều kiện	margin đủ	IN_POSITION	đặt lệnh
IN_POSITION	giá chạm SL	—	EXIT	đóng lệnh
EXIT	đã đóng	—	COOLDOWN	chờ
COOLDOWN	hết giờ	—	IDLE	—

2. Điều kiện chuyển trạng thái (Guard Conditions)

• Điều kiện phải đo lường được: close > ma20 and adx > 25 and spread < 0.5.
• Cổng (guard) ngăn chuyển trạng thái khi chưa đủ điều kiện.
• Ưu tiên xử lý sự kiện thoát (SL/TP) trước sự kiện vào lệnh.

2.1. Ví dụ guard

def can_enter(self):
    return (self.spread < self.max_spread and
            self.adx > 25 and
            self.margin_free > self.required_margin)

3. Xử lý tình huống biên (Edge Cases)

3.1. Mất kết nối

Giữ nguyên trạng thái, chờ reconnect. Sau khi nối lại, đối soát trạng thái thực tế trên sàn (lệnh đang mở?) rồi đồng bộ lại FSM. Không vào lệnh mới trong lúc mất kết nối.

3.2. Lỗi đặt lệnh

Nếu lệnh bị từ chối (spread rộng, margin thiếu), quay về trạng thái trước + bộ đếm thử lại, tránh lặp vô hạn làm nghẽn API.

3.3. Bot restart / VPS restart

Lưu trạng thái vào Global Variable (MT5) hoặc file/DB. Khi khởi động, đọc lại và tiếp tục đúng vòng đời.

3.4. Tin tức lớn

Có trạng thái/flag NEWS_BLOCK: trước tin lớn, chặn vào lệnh mới nhưng vẫn quản lý lệnh đang mở.

4. Code FSM đầy đủ có guard và timeout

import time

class BotFSM:
    def __init__(self):
        self.state = "IDLE"
        self.retries = 0
        self.timer = time.time()

    def on_tick(self, bid, ask, signal, adx, spread):
        if self.state == "IDLE":
            if signal == "BUY" and adx > 25 and spread < 0.5:
                self.state = "ENTER"
        elif self.state == "ENTER":
            if self.place_order("BUY", ask):
                self.state = "IN_POSITION"
                self.retries = 0
            else:
                self.retries += 1
                if self.retries > 3:
                    self.state = "IDLE"   # từ bỏ, chờ tín hiệu mới
        elif self.state == "IN_POSITION":
            if bid <= self.stop_price:
                self.close_all()
                self.state = "EXIT"
                self.timer = time.time()
        elif self.state == "EXIT":
            if time.time() - self.timer > self.cooldown:
                self.state = "IDLE"

    def place_order(self, side, price): return True
    def close_all(self): ...

5. FSM nâng cao: đa cụm và điều phối

Bot hedging rải nhiều cụm; mỗi cụm có FSM riêng, bot mẹ điều phối qua global state:

for c in clusters:
    c.on_tick(bid, ask)         # mỗi cụm tự chuyển trạng thái
sync_global_state(clusters)     # cập nhật trạng thái tổng

Cụm này SURPLUS (chốt lời) trong khi cụm khác HEDGE (phòng thủ) — tất cả đều tuân theo FSM riêng, tránh xung đột lệnh.

6. Kiểm thử FSM

1. Duyệt bảng chuyển trạng thái: mọi (state, event) hợp lệ.
2. Mô phỏng mất kết nối, lỗi lệnh, tin lớn, spread rộng, restart.
3. Chạy backtest, xem log chuyển trạng thái từng tick.
4. Paper trade 2–4 tuần trước real.

7. Sai lầm thường gặp

• Quên cập nhật state sau hành động.
• Thiếu COOLDOWN → nhồi lệnh.
• Không lưu state khi restart.
• Chuyển trạng thái bất hợp lệ (COOLDOWN → ENTER thẳng).
• Không log → khó truy vết.

8. Câu hỏi thường gặp

FSM khác gì vòng lặp if/else? FSM buộc bạn tư duy theo trạng thái + sự kiện, dễ quản lý và mở rộng.

Bao nhiêu trạng thái là đủ? Bắt đầu 5–6, thêm khi cần. Đừng thiết kế quá phức tạp ban đầu.

Có cần state riêng cho từng symbol không? Nếu bot giao dịch nhiều symbol, nên có FSM riêng mỗi symbol (hoặc map theo symbol).

6.1. Khai báo bảng chuyển trạng thái dạng dữ liệu — máy tự kiểm tra

Với bot phức tạp, hãy khai báo bảng chuyển trạng thái dưới dạng dữ liệu để máy tự kiểm tra — tránh lỗi do code lệch với thiết kế:

TRANSITIONS = {
    ("IDLE", "BUY_SIGNAL"):  ("ENTER", "check_filters"),
    ("ENTER", "OK"):         ("IN_POSITION", "place_order"),
    ("ENTER", "FAIL"):       ("IDLE", "log_retry"),
    ("IN_POSITION", "SL"):   ("EXIT", "close_all"),
    ("IN_POSITION", "TP"):   ("EXIT", "close_all"),
    ("EXIT", "DONE"):        ("COOLDOWN", "set_timer"),
    ("COOLDOWN", "TIMEOUT"): ("IDLE", "reset"),
}

def transition(current, event, **ctx):
    key = (current, event)
    if key not in TRANSITIONS:
        raise ValueError(f"Chuyển trạng thái không hợp lệ: {key}")
    nxt, action = TRANSITIONS[key]
    getattr(bot, action)(**ctx)
    bot.state = nxt
    log_transition(current, event, nxt)

Mọi chuyển trạng thái bất hợp lệ sẽ ném lỗi ngay khi dev — thay vì âm thầm sai lúc chạy real.

6.2. Timer / timeout tuyệt đối

Trạng thái có thời gian chờ (ENTER, EXIT) cần timeout để bot không kẹt vô hạn khi API không phản hồi:

import time

class TimerFSM(BotFSM):
    def __init__(self, timeout=10):
        super().__init__()
        self.timeout = timeout
        self.entered_at = time.time()

    def _tick(self):
        if self.state in ("ENTER", "EXIT"):
            if time.time() - self.entered_at > self.timeout:
                log("Timeout - quay về IDLE")
                self.state = "IDLE"

6.3. Log mọi transition kèm chi tiết (telemetry)

import datetime

def log_transition(frm, event, to, extra=""):
    line = f"{datetime.datetime.now()} | {frm} --{event}--> {to} | {extra}"
    with open("fsm.log", "a", encoding="utf-8") as f:
        f.write(line + "
")

# ví dụ: log_transition("IDLE", "BUY_SIGNAL", "ENTER", f"adx={adx} spread={spread}")

Sau một tuần chạy, bạn dùng log này để dựng biểu đồ “bot ở trạng thái nào bao lâu” — phát hiện vòng lặp kẹt (ví dụ cứ ENTER rồi FAIL liên tục) mà không cần nhìn màn hình.

6.4. Nested FSM cho bot đa symbol

Khi quản lý nhiều symbol cùng lúc, mỗi symbol một FSM con và một FSM mẹ quản lý tổng thể:

class SymbolFSM:      # FSM con cho từng symbol
    def __init__(self, symbol):
        self.symbol = symbol
        self.state = "IDLE"

class MasterFSM:      # FSM mẹ
    def __init__(self):
        self.symbols = {s: SymbolFSM(s) for s in ("XAUUSD", "BTCUSDT")}

    def on_tick(self, data):
        for sym, fsm in self.symbols.items():
            fsm.on_tick(data[sym])     # mỗi con tự chạy
        self.risk_check()              # mẹ kiểm soát rủi ro tổng

FSM mẹ đảm bảo tổng rủi ro toàn danh mục không vượt ngưỡng, trong khi mỗi con xử lý tín hiệu riêng cho symbol của nó.

Ví dụ FSM hoàn chỉnh có cảnh báo Telegram

import time, requests

class GridFSM:
    def __init__(self):
        self.state = "IDLE"
        self.orders = []
        self.cooldown_until = 0
        self.telegram = "https://api.telegram.org/botXXX/sendMessage"

    def _alert(self, msg):
        requests.post(self.telegram,
                      json={"chat_id": "123", "text": f"[{self.state}] {msg}"},
                      timeout=5)

    def on_tick(self, bid, ask):
        if self.state == "IDLE":
            if self.entry_signal():
                self._alert("Vào lệnh cụm đầu")
                self.state = "GRID"
        elif self.state == "GRID":
            if bid < self.last() - self.step and len(self.orders) < self.max_grid:
                self.orders.append(ask)
                self._alert(f"Thêm bậc {len(self.orders)}")
            if self.profit_ready():
                self._alert("Chốt lời cụm")
                self.state = "EXIT"
        elif self.state == "EXIT":
            if not self.orders:
                self.state = "COOLDOWN"
        elif self.state == "COOLDOWN":
            if time.time() > self.cooldown_until:
                self.state = "IDLE"

Mỗi lần đổi trạng thái, bot gửi cảnh báo Telegram — bạn biết bot “đang làm gì” ngay cả khi không mở máy tính.

Kiểm tra FSM bằng bảng test tự động

CASES = [
    ("IDLE", "BUY_SIGNAL"), ("ENTER", "OK"), ("IN_POSITION", "SL"),
    ("EXIT", "DONE"), ("COOLDOWN", "TIMEOUT"),
]
for state, event in CASES:
    nxt = transition(state, event)
    print(f"{state} + {event} -> {nxt}")
    assert nxt is not None, f"Thiếu transition {state}/{event}"

Chạy test này mỗi khi sửa code — nếu quên khai báo một chuyển trạng thái, test báo ngay lập tức.

Đo hiệu suất FSM: đếm số transition mỗi ngày

Đếm số transition để phát hiện bot “nhảy loạn” (vào/thoát liên tục — dấu hiệu nhiễu tín hiệu):

from collections import Counter

trans = Counter()
def log_transition(frm, event, to):
    trans[(frm, to)] += 1

# cuối ngày:
for (frm, to), n in trans.most_common(10):
    print(f"{frm} -> {to}: {n} lần")

Nếu thấy ENTER → FAIL lặp hàng trăm lần, bạn biết ngay bộ lọc tín hiệu đang quá lỏng và cần siết điều kiện.

Bảng chuyển trạng thái cho từng loại chiến lược cụ thể

Không phải mọi bot đều dùng cùng một bộ trạng thái. Dưới đây là bảng gợi ý cho ba chiến lược phổ biến:

1. Bot lưới (Grid)

Trạng thái	Ý nghĩa	Sự kiện chuyển
IDLE	Chờ tín hiệu khởi động lưới	tín hiệu → GRID_OPEN
GRID_OPEN	Lưới đang mở, thêm bậc theo step	giá hồi về vùng lời → SURPLUS; chạm max grid → GRID_FULL
GRID_FULL	Đạt tối đa số bậc, dừng thêm	giá hồi → SURPLUS
SURPLUS	Chốt lời từng bậc	chốt hết → EXIT
EXIT	Đóng toàn bộ cụm	đóng xong → COOLDOWN
COOLDOWN	Nghỉ ngắn	hết giờ → IDLE

2. Bot hedging

Trạng thái	Ý nghĩa	Sự kiện chuyển
IDLE	Chờ tín hiệu	tín hiệu → ENTER
ENTER	Mở lệnh đầu cụm	khớp → IN_POSITION
IN_POSITION	Đang giữ lệnh, theo dõi	lỗ sâu → HEDGE; hồi về lời → SURPLUS
HEDGE	Mở lệnh đối ứng khóa rủi ro	giá ổn định → UNWIND
UNWIND	Dỡ dần lệnh hedge	xong → SURPLUS / EXIT
SURPLUS	Chốt lời	hết → EXIT

3. Bot theo xu hướng (trend following)

Trạng thái	Ý nghĩa	Sự kiện chuyển
SCAN	Quét tín hiệu trên nhiều khung	tín hiệu đủ điều kiện → CONFIRM
CONFIRM	Xác nhận với khung lớn	đồng thuận → ENTER; không → SCAN
ENTER	Vào lệnh theo xu hướng	khớp → IN_TREND
IN_TREND	Giữ lệnh, trailing	đảo chiều → EXIT; chạm TP → EXIT
EXIT	Đóng lệnh	xong → SCAN

Mấu chốt: thiết kế trạng thái theo logic chiến lược, không theo cảm tính. Mỗi trạng thái phải trả lời được câu hỏi “bot đang làm gì và khi nào chuyển tiếp?”.

State Explosion — khi FSM quá phức tạp

Khi bot đa cụm, mỗi cụm nhiều trạng thái, tổng tổ hợp trạng thái có thể bùng nổ. Ví dụ 5 cụm × 6 trạng thái = 7.776 tổ hợp lý thuyết. Cách quản lý:

• Tách FSM con: mỗi cụm một FSM riêng, không nhét tất cả vào một máy trạng thái khổng lồ.
• Chỉ liệt kê transition hợp lệ: dùng bảng (state, event) → state, máy tự chặn transition bất hợp lệ.
• Trạng thái tổng (aggregate): bot mẹ chỉ quan tâm tóm tắt (đang an toàn / cần can thiệp), không cần biết chi tiết từng cụm.
• Tránh trạng thái thừa: nếu hai trạng thái có hành vi giống hệt, gộp làm một.

Xử lý ngoại lệ trong FSM — bot không được “chết im lặng”

Lỗi runtime (mất kết nối, API trả null, chia cho 0) có thể làm bot crash giữa chừng. FSM chuyên nghiệp phải bọc toàn bộ vòng lặp:

def safe_tick(self, bid, ask):
    try:
        self.on_tick(bid, ask)
    except Exception as e:
        log(f"Lỗi trong on_tick: {e}")
        # chuyển trạng thái an toàn
        if self.state in ("ENTER", "EXIT"):
            self.state = "IDLE"
        # đếm lỗi liên tiếp để tránh vòng lặp chết
        self.error_count += 1
        if self.error_count > 10:
            self.state = "PAUSED"   # dừng hẳn, chờ con người

Trạng thái PAUSED rất quan trọng: nếu bot lỗi liên tục, nó phải tự dừng chứ không cố gắng giao dịch trong tình trạng hỏng hóc — thà không giao dịch còn hơn giao dịch sai.

Đồng bộ trạng thái sau restart — quy trình chi tiết

Khi VPS restart hoặc bot khởi động lại, phải khôi phục đúng trạng thái:

1. Đọc trạng thái đã lưu (Global Variable / JSON / DB).
2. Đối soát với sàn: lấy danh sách lệnh thực tế theo magic.
3. Khôi phục: nếu FSM nói IN_POSITION nhưng sàn không còn lệnh → chuyển EXIT/COOLDOWN. Nếu FSM nói IDLE nhưng sàn có lệnh bot → chuyển IN_POSITION (lệnh sót từ trước khi lưu state).
4. Ghi log khởi động: ghi rõ trạng thái phục hồi để dễ truy vết.

def restore(bot):
    saved = load_state()
    real_orders = get_orders_by_magic(bot.magic)
    if saved.state == "IN_POSITION" and not real_orders:
        bot.state = "COOLDOWN"
        log("Phục hồi: lệnh không còn trên sàn")
    elif saved.state == "IDLE" and real_orders:
        bot.state = "IN_POSITION"
        log("Phục hồi: phát hiện lệnh sót, quản lý tiếp")
    else:
        bot.state = saved.state

Đo hiệu quả FSM — số liệu vận hành

Để biết FSM hoạt động tốt hay không, theo dõi các số liệu:

• % thời gian ở mỗi trạng thái: bot nên ở IN_POSITION phần lớn thời gian khi có xu hướng; nếu cứ IDLE mãi thì tín hiệu quá khắt khe.
• Số transition/ngày: quá nhiều (hàng nghìn) = bot nhảy loạn do nhiễu; quá ít = bỏ lỡ cơ hội.
• Tỷ lệ ENTER→FAIL: cao nghĩa là điều kiện vào chưa đủ (spread, margin).
• Thời gian trung bình 1 chu kỳ lệnh: từ IDLE→EXIT nên phù hợp với chiến lược (vài giờ–vài ngày).

Câu hỏi thường gặp nâng cao về logic FSM

Làm sao xử lý nhiều symbol cùng lúc? Mỗi symbol một instance FSM (dict theo symbol). FSM mẹ quản lý rủi ro tổng.

FSM có nên dùng thread/async? Với MT5 API thường là 1 luồng tuần tự đủ. Với nhiều symbol muốn song song thì dùng asyncio, mỗi FSM con là một task.

Khi nào chuyển từ FSM sang Behavior Tree? Khi có quá nhiều mục tiêu xung đột (vừa đuổi lợi nhuận vừa bảo toàn vốn vừa đa chiến lược) — Behavior Tree xử lý ưu tiên tốt hơn.

State machine của MT5 (OnTick + GlobalVariable) có phải FSM không? Đó là cơ chế lưu trạng thái, không phải FSM. FSM là cách tổ chức logic chuyển trạng thái; MT5 cung cấp nơi lưu, bạn tự xây logic.

Thiết kế FSM cho bot đa cụm hedging — điều phối chi tiết

Bot hedging đa cụm là bài toán phức tạp nhất: nhiều cụm lệnh, mỗi cụm có vòng đời riêng, nhưng chúng phải phối hợp để không phá vỡ rủi ro tổng. Kiến trúc đề xuất:

class ClusterFSM:
    # FSM của một cụm lệnh
    def __init__(self, cid):
        self.cid = cid
        self.state = "IDLE"
        self.orders = []
        self.total_loss = 0

    def on_tick(self, bid, ask, global_ctx):
        if self.state == "IDLE":
            if global_ctx.risk_ok() and self.signal():
                self.state = "ENTER"
        elif self.state == "ENTER":
            self.open_first()
            self.state = "IN_POSITION"
        elif self.state == "IN_POSITION":
            if self.total_loss < -global_ctx.hedge_threshold:
                self.state = "HEDGE"
            elif self.surplus_ready():
                self.state = "SURPLUS"
        elif self.state == "HEDGE":
            self.manage_hedge()
            if self.safe_zone():
                self.state = "UNWIND"
        elif self.state == "UNWIND":
            if not self.hedge_orders:
                self.state = "SURPLUS"
        elif self.state == "SURPLUS":
            if not self.orders:
                self.state = "EXIT"
        elif self.state == "EXIT":
            if not self.orders:
                self.state = "COOLDOWN"

Bot mẹ (MasterFSM) duyệt tất cả cụm mỗi tick, gọi on_tick của từng cụm và kiểm tra rủi ro tổng:

class MasterFSM:
    def __init__(self):
        self.clusters = {}
        self.hedge_threshold = -500   # USD

    def on_tick(self, bid, ask):
        total_loss = sum(c.total_loss for c in self.clusters.values())
        ctx = SimpleNamespace(risk_ok=lambda: total_loss > self.hedge_threshold,
                              hedge_threshold=self.hedge_threshold)
        for c in self.clusters.values():
            c.on_tick(bid, ask, ctx)

Ưu điểm: mỗi cụm tự quyết định cục bộ theo FSM riêng, còn bot mẹ đảm bảo ràng buộc toàn cục (tổng lỗ không vượt ngưỡng). Đây chính là mô hình PyBot thực chiến sử dụng.

Ví dụ log chuyển trạng thái thực tế và cách đọc

2026-08-14 09:00:01 [IDLE] tick bid=2350.0 ask=2350.3
2026-08-14 09:07:33 [IDLE --BUY_SIGNAL--> ENTER] adx=28 spread=0.21
2026-08-14 09:07:34 [ENTER --OK--> IN_POSITION] margin=1500
2026-08-14 09:07:35 [IN_POSITION] open BUY 0.01 @ 2350.5 sl=2343 tp=2364
2026-08-14 11:30:00 [IN_POSITION --TP--> EXIT] pnl=+13.5
2026-08-14 11:30:01 [EXIT --DONE--> COOLDOWN] timer=120s
2026-08-14 11:32:01 [COOLDOWN --TIMEOUT--> IDLE]

Khi đọc log, chú ý: chuỗi transition có hợp lý không? Có transition “nhảy cóc” (vd IDLE thẳng EXIT) là dấu hiệu bug. Thời gian ở mỗi trạng thái có hợp lý với chiến lược không? Lệnh mở có đúng magic, đúng lot, đúng SL/TP không?

So sánh FSM tự viết vs dùng thư viện

Tiêu chí	Tự viết	Thư viện (vd transitions)
Kiểm soát	Toàn quyền	Theo API thư viện
Khối lượng code	Nhiều hơn	Ít hơn, khai báo gọn
Học được nguyên lý	Sâu	Đủ dùng
Debug	Dễ kiểm soát	Tùy thư viện
Khuyến nghị	Học tập, kiểm soát tối đa	Sản xuất nhanh

Khuyến nghị: học bằng cách tự viết 1–2 lần để hiểu nguyên lý, sau đó dùng thư viện cho dự án thật để tiết kiệm thời gian và tránh lỗi thủ công.

Câu hỏi thường gặp bổ sung

FSM có xử lý được lệnh treo (pending order) không? Có — thêm trạng thái PENDING cho các lệnh chờ kích hoạt (limit/stop), chuyển sang IN_POSITION khi kích hoạt.

Nên lưu trạng thái ở đâu cho bot đa cụm? Mỗi cụm một record trong JSON/DB, có cluster_id, state, danh sách lệnh, tổng lỗ. Bot mẹ lưu tóm tắt riêng.

Làm sao test FSM đa cụm? Mô phỏng nhiều kịch bản: một cụm SURPLUS trong khi cụm khác HEDGE, tổng lỗ chạm ngưỡng, một cụm mất kết nối. Dựng log và kiểm tra ràng buộc toàn cục không bị vi phạm.

Kiểm thử FSM với dữ liệu thật — mô phỏng tick

Test đơn vị mới chỉ kiểm tra logic. Để chắc chắn bot hoạt động trong thực tế, hãy mô phỏng một chuỗi tick thật và quan sát toàn bộ vòng đời:

import random, time

def simulate(bot, n_ticks=5000):
    price = 67000.0
    log = []
    for i in range(n_ticks):
        price += random.gauss(0, 20)   # bước giá ngẫu nhiên
        bid, ask = price, price + 0.3
        before = bot.state
        bot.on_tick(bid, ask)
        if bot.state != before:
            log.append((i, before, bot.state, round(bid, 1)))
    return log

bot = BotFSM()
events = simulate(bot)
print(f"Tổng transition: {len(events)}")
for e in events[:20]:
    print(f"  tick {e[0]}: {e[1]} -> {e[2]} @ {e[3]}")

Chạy mô phỏng nhiều lần với seed ngẫu nhiên khác nhau, kiểm tra: bot không bao giờ vào lệnh khi đang IN_POSITION, không bao giờ kẹt ở một trạng thái quá lâu, và mọi transition đều nằm trong bảng cho phép.

Kiến trúc khuyến nghị cho dự án bot FSM

Khi đưa FSM vào dự án thật, hãy tách biệt rõ ràng các lớp để dễ bảo trì:

my_bot/
├─ fsm/
│  ├─ core.py          # lớp FSM cơ bản + bảng transition
│  ├─ states.py        # định nghĩa các trạng thái
│  └─ transitions.py   # bảng chuyển trạng thái dạng dữ liệu
├─ brokers/
│  └─ mt5.py           # kết nối MT5, đặt lệnh
├─ strategies/
│  ├─ grid.py          # chiến lược lưới
│  └─ hedge.py         # chiến lược hedging
├─ storage/
│  └─ state_store.py   # lưu/đọc trạng thái
└─ main.py             # khởi động, vòng lặp chính

Cấu trúc này giúp bạn đổi chiến lược mà không đụng vào FSM core, và đổi broker mà không đụng vào chiến lược — mỗi thay đổi giới hạn trong một file.

9. Kết luận

Logic chuyển trạng thái quyết định bot “đúng người đúng việc”. Thiết kế bảng chuyển trạng thái trước, viết code sau, luôn lưu state + log và kiểm thử kỹ mọi đường chuyển.

📌 Thực hành FSM chi tiết trong khóa Python Hedging / Bot Auto Trading tại Hướng Nghiệp Dữ Liệu.