โบนัสสำหรับนักพัฒนา — Python Pseudocode สำหรับกลยุทธ์ SMC ทั้ง 18 รูปแบบ/ICT

บทเรียนที่ 11 จัดอันดับกลยุทธ์ทั้ง 18 อย่าง โดยใช้ความแม่นยำ บทเรียนนี้จะแปลงกลยุทธ์ทั้งหมดให้เป็น Python — กฎเชิงกลไกที่แน่นอนที่อธิบายไว้ในคำจำกัดความของแต่ละกลยุทธ์ ซึ่งแสดงเป็นฟังก์ชันที่คุณสามารถอ่าน ปรับเปลี่ยน หรือวางลงใน backtester ได้

รูปแบบข้อมูล

ทุกฟังก์ชันคาดหวัง bars: รายการของ dicts เรียงตามลำดับจากเก่าสุดไปยังใหม่สุด

# bars[i] = {"time": "2026-07-01", "open": 100.0, "high": 101.0, "low": 99.5, "close": 100.5}

กลยุทธ์ที่ขึ้นอยู่กับเวลา — Silver Bullet, NY AM Session Reversal — ต้องการค่า "time" ที่รวมชั่วโมง:นาที เช่น "2026-07-01 10:15" แท่งข้อมูล "YYYY-MM-DD" ธรรมดาไม่มีข้อมูล killzone และฟังก์ชันทั้งสองจะคืนค่า None บนข้อมูลรายวัน

Primitives ที่ใช้ร่วมกัน

แปดฟังก์ชันที่นำกลับมาใช้ใหม่ได้ หนึ่งฟังก์ชันต่อแนวคิดหลักในบทเรียน 1–4/7/8 ทุกกลยุทธ์ด้านล่างสร้างจากสิ่งเหล่านี้ — อ่านส่วนนี้ครั้งเดียว และฟังก์ชันทั้ง 18 ที่ตามมาจะเป็นการประกอบ ไม่ใช่ตรรกะใหม่

from dataclasses import dataclass
from typing import Optional


@dataclass
class Swing:
    index: int
    price: float
    kind: str  # "high" or "low"


@dataclass
class Signal:
    strategy: str
    tier: int
    direction: str  # "long" or "short"
    entry: float
    stop: Optional[float]
    target: Optional[float] = None
    note: str = ""


def find_swings(bars, lookback=2):
    """Simple fractal swing detector: a bar is a swing high/low if it's the
    max/min within `lookback` bars on each side. Illustrative only — real
    swing detection needs noise filtering this doesn't attempt."""
    swings = []
    n = len(bars)
    for i in range(lookback, n - lookback):
        window = bars[i - lookback : i + lookback + 1]
        highs = [b["high"] for b in window]
        lows = [b["low"] for b in window]
        if bars[i]["high"] == max(highs):
            swings.append(Swing(i, bars[i]["high"], "high"))
        if bars[i]["low"] == min(lows):
            swings.append(Swing(i, bars[i]["low"], "low"))
    return swings


def label_structure(swings):
    """Classify each swing as HH/HL/LH/LL relative to the prior swing of the
    same kind, then flag BOS (break of structure) / CHoCH (change of
    character). Returns a list of (swing, label, break_type) tuples."""
    labeled = []
    last_high = last_low = None
    trend = None  # "up" or "down", once established
    for s in swings:
        if s.kind == "high":
            label = "HH" if last_high and s.price > last_high.price else ("LH" if last_high else "H")
            last_high = s
        else:
            label = "HL" if last_low and s.price > last_low.price else ("LL" if last_low else "L")
            last_low = s
        break_type = None
        if trend == "up" and label == "LL":
            break_type, trend = "CHoCH", "down"
        elif trend == "down" and label == "HH":
            break_type, trend = "CHoCH", "up"
        elif trend == "up" and label == "HH":
            break_type = "BOS"
        elif trend == "down" and label == "LL":
            break_type = "BOS"
        elif trend is None and label in ("HH", "LL"):
            trend = "up" if label == "HH" else "down"
        labeled.append((s, label, break_type))
    return labeled


def find_liquidity_pools(swings, equal_tolerance=0.05):
    """Group swings of the same kind within `equal_tolerance` of each other
    into liquidity pools (EQH/EQL clusters, and standalone old highs/lows)."""
    pools = []
    for kind in ("high", "low"):
        same = [s for s in swings if s.kind == kind]
        used = set()
        for i, s in enumerate(same):
            if i in used:
                continue
            cluster = [s]
            for j in range(i + 1, len(same)):
                if j in used:
                    continue
                if abs(same[j].price - s.price) <= equal_tolerance:
                    cluster.append(same[j])
                    used.add(j)
            pools.append({"kind": kind, "price": sum(c.price for c in cluster) / len(cluster), "swings": cluster})
    return pools


def detect_sweep(bars, pool, confirm_bars=1):
    """A sweep = price trades through the pool level, then closes back on
    the other side within `confirm_bars`. Returns the sweep bar index or None."""
    for i, bar in enumerate(bars):
        if pool["kind"] == "low" and bar["low"] < pool["price"]:
            for j in range(i, min(i + 1 + confirm_bars, len(bars))):
                if bars[j]["close"] > pool["price"]:
                    return i
        if pool["kind"] == "high" and bar["high"] > pool["price"]:
            for j in range(i, min(i + 1 + confirm_bars, len(bars))):
                if bars[j]["close"] < pool["price"]:
                    return i
    return None


def find_order_block(bars, break_index, direction):
    """The order block is the last opposite-direction candle before
    `break_index`. direction="bullish" looks for the last down-candle,
    "bearish" the last up-candle."""
    for i in range(break_index - 1, -1, -1):
        bar = bars[i]
        is_down = bar["close"] < bar["open"]
        is_up = bar["close"] > bar["open"]
        if direction == "bullish" and is_down:
            return {"low": bar["low"], "high": bar["high"], "index": i}
        if direction == "bearish" and is_up:
            return {"low": bar["low"], "high": bar["high"], "index": i}
    return None


def find_fvg(bars):
    """3-candle Fair Value Gap scan: candle[i]'s wick doesn't overlap
    candle[i+2]'s wick. Returns a list of gap dicts."""
    gaps = []
    for i in range(len(bars) - 2):
        c1, c3 = bars[i], bars[i + 2]
        if c1["high"] < c3["low"]:
            gaps.append({"kind": "bullish", "low": c1["high"], "high": c3["low"], "index": i + 1})
        if c1["low"] > c3["high"]:
            gaps.append({"kind": "bearish", "low": c3["high"], "high": c1["low"], "index": i + 1})
    return gaps


def premium_discount_zone(dealing_high, dealing_low, price):
    """Returns premium/discount relative to the dealing range midpoint, plus
    the OTE (62-79%) band bounds for a bullish retracement."""
    midpoint = (dealing_high + dealing_low) / 2
    zone = "premium" if price > midpoint else "discount"
    rng = dealing_high - dealing_low
    return {
        "zone": zone,
        "midpoint": midpoint,
        "ote_high": dealing_high - rng * 0.62,
        "ote_low": dealing_high - rng * 0.79,
    }


def _parse_hour_minute(time_str):
    """Extract (hour, minute) from a 'YYYY-MM-DD HH:MM' timestamp. Returns
    (None, None) for a plain daily bar."""
    if " " not in time_str:
        return None, None
    _, clock = time_str.split(" ", 1)
    h, m = clock.split(":")
    return int(h), int(m)


def in_killzone(hour, minute, window):
    """window: (start_hour, start_min, end_hour, end_min), NY time."""
    start_h, start_m, end_h, end_m = window
    t = hour * 60 + minute
    return (start_h * 60 + start_m) <= t <= (end_h * 60 + end_m)

Tier 1 — เชิงกลไกสูง

OTE + Order Block Confluence

def ote_order_block_confluence(bars):
    """Tier 1. Long entry when a bullish order block's range overlaps the
    62-79% OTE band of the same swing. Precise: the overlap either exists
    or it doesn't."""
    swings = find_swings(bars)
    if len(swings) < 2:
        return None
    dealing_high = max(s.price for s in swings if s.kind == "high")
    dealing_low = min(s.price for s in swings if s.kind == "low")
    pd = premium_discount_zone(dealing_high, dealing_low, bars[-1]["close"])
    high_indices = [s.index for s in swings if s.kind == "high"]
    if not high_indices:
        return None
    ob = find_order_block(bars, max(high_indices), "bullish")
    if not ob:
        return None
    overlap = max(ob["low"], pd["ote_low"]) <= min(ob["high"], pd["ote_high"])
    if overlap and bars[-1]["low"] <= ob["high"] and bars[-1]["close"] > ob["low"]:
        return Signal("OTE + Order Block Confluence", 1, "long", bars[-1]["close"], ob["low"] * 0.995, dealing_high)
    return None

Break-and-Retest

def break_and_retest(bars):
    """Tier 1. Long entry: BOS confirmed, then price retests the broken
    level and holds."""
    labeled = label_structure(find_swings(bars))
    bos_events = [s for s, _, brk in labeled if brk == "BOS"]
    if not bos_events:
        return None
    broken_level = bos_events[-1].price
    last = bars[-1]
    if last["low"] <= broken_level <= last["high"] and last["close"] > broken_level:
        return Signal("Break-and-Retest", 1, "long", last["close"], broken_level * 0.995)
    return None

Turtle Soup

def turtle_soup(bars, lookback=20):
    """Tier 1. A false breakout of a prior high/low that reverses within
    the same or next bar."""
    if len(bars) < lookback + 2:
        return None
    prior_low = min(b["low"] for b in bars[-lookback - 2 : -2])
    trigger, confirm = bars[-2], bars[-1]
    if trigger["low"] < prior_low and confirm["close"] > prior_low:
        return Signal("Turtle Soup", 1, "long", confirm["close"], trigger["low"] * 0.995)
    return None

Silver Bullet

def silver_bullet(bars, killzone=(10, 0, 11, 0)):
    """Tier 1. An FVG forming inside the 10-11am NY window. Requires
    intraday bars — see _parse_hour_minute."""
    gaps = find_fvg(bars)
    if not gaps:
        return None
    gap = gaps[-1]
    hour, minute = _parse_hour_minute(bars[gap["index"]]["time"])
    if hour is None or not in_killzone(hour, minute, killzone):
        return None
    direction = "long" if gap["kind"] == "bullish" else "short"
    entry = gap["high"] if direction == "long" else gap["low"]
    stop = gap["low"] * 0.995 if direction == "long" else gap["high"] * 1.005
    return Signal("Silver Bullet", 1, direction, entry, stop)

2022 Model

def model_2022(bars):
    """Tier 1. Sweep of an old high/low → CHoCH confirmation → entry on the
    retracement into the resulting FVG."""
    swings = find_swings(bars)
    pools = find_liquidity_pools(swings)
    labeled = label_structure(swings)
    choch = [s for s, _, brk in labeled if brk == "CHoCH"]
    if not choch:
        return None
    kind = "low" if choch[-1].kind == "low" else "high"
    matching_pools = [p for p in pools if p["kind"] == kind]
    if not matching_pools:
        return None
    sweep_index = detect_sweep(bars, matching_pools[-1])
    if sweep_index is None:
        return None
    gaps = find_fvg(bars[sweep_index:])
    if not gaps:
        return None
    gap = gaps[-1]
    direction = "long" if gap["kind"] == "bullish" else "short"
    stop = gap["low"] * 0.995 if direction == "long" else gap["high"] * 1.005
    return Signal("2022 Model", 1, direction, bars[-1]["close"], stop)

Tier 2 — แม่นยำปานกลาง

Judas Swing

def judas_swing(bars, session_open_index=0):
    """Tier 2. An early false move at the session open sweeps liquidity
    before the real move begins."""
    window = bars[session_open_index : session_open_index + 3]
    if len(window) < 3:
        return None
    first, mid, last = window
    if mid["low"] < first["low"] and last["close"] > first["open"]:
        return Signal("Judas Swing", 2, "long", last["close"], mid["low"] * 0.995)
    return None

Unicorn Model

def unicorn_model(bars):
    """Tier 2. A breaker block and a fair value gap overlapping at the same
    zone — higher confluence than either alone."""
    gaps = find_fvg(bars)
    labeled = label_structure(find_swings(bars))
    for s, label, brk in labeled:
        if brk != "CHoCH":
            continue
        direction_ob = "bullish" if label == "HL" else "bearish"
        ob = find_order_block(bars, s.index, direction_ob)
        if not ob:
            continue
        for gap in gaps:
            if max(ob["low"], gap["low"]) <= min(ob["high"], gap["high"]):
                direction = "long" if direction_ob == "bullish" else "short"
                stop = ob["low"] * 0.995 if direction == "long" else ob["high"] * 1.005
                return Signal("Unicorn Model", 2, direction, bars[-1]["close"], stop)
    return None

Breaker Block Reversal

def breaker_block_reversal(bars):
    """Tier 2. A failed order block flips polarity and is traded from the
    other side."""
    labeled = label_structure(find_swings(bars))
    for s, _, brk in labeled:
        if brk != "BOS":
            continue
        ob = find_order_block(bars, s.index, "bullish")
        if not ob:
            continue
        failed = any(b["close"] < ob["low"] for b in bars[s.index:])
        if not failed:
            continue
        retest = bars[-1]
        if ob["low"] <= retest["high"] <= ob["high"] and retest["close"] < ob["low"]:
            return Signal("Breaker Block Reversal", 2, "short", retest["close"], ob["high"] * 1.005)
    return None

Mitigation Block Entry

def mitigation_block_entry(bars, narrow_window=2):
    """Tier 2. Entry on a narrow pre-launch consolidation rather than the
    full order block range."""
    for i in range(narrow_window, len(bars) - 1):
        consolidation = bars[i - narrow_window : i]
        launch = bars[i]
        cons_high = max(b["high"] for b in consolidation)
        cons_low = min(b["low"] for b in consolidation)
        if (launch["close"] - launch["open"]) <= 2 * (cons_high - cons_low):
            continue  # not a launch candle
        last = bars[-1]
        if cons_low <= last["low"] <= cons_high and last["close"] > cons_low:
            return Signal("Mitigation Block Entry", 2, "long", last["close"], cons_low * 0.995)
    return None

Inversion FVG Reversal

def inversion_fvg_reversal(bars):
    """Tier 2. A fair value gap that price closes back through completely
    flips from support to resistance (or the reverse)."""
    for gap in find_fvg(bars):
        closed_through = any(
            (b["close"] < gap["low"] if gap["kind"] == "bullish" else b["close"] > gap["high"])
            for b in bars[gap["index"] + 1 :]
        )
        if not closed_through:
            continue
        last = bars[-1]
        if gap["low"] <= last["high"] <= gap["high"]:
            direction = "short" if gap["kind"] == "bullish" else "long"
            stop = gap["high"] * 1.005 if direction == "short" else gap["low"] * 0.995
            return Signal("Inversion FVG Reversal", 2, direction, last["close"], stop)
    return None

NY AM Session Reversal

def ny_am_session_reversal(bars, killzone=(10, 0, 11, 0)):
    """Tier 2. Same mechanics as a plain sweep-and-reverse, filtered to only
    count inside the NY AM killzone. Requires intraday bars."""
    pools = find_liquidity_pools(find_swings(bars))
    low_pools = [p for p in pools if p["kind"] == "low"]
    if not low_pools:
        return None
    sweep_index = detect_sweep(bars, low_pools[-1])
    if sweep_index is None:
        return None
    hour, minute = _parse_hour_minute(bars[sweep_index]["time"])
    if hour is None or not in_killzone(hour, minute, killzone):
        return None
    return Signal("NY AM Session Reversal", 2, "long", bars[-1]["close"], low_pools[-1]["price"] * 0.995)

Liquidity Void Fill

def liquidity_void_fill(bars, body_ratio=2.5):
    """Tier 2. A thin, fast candle's range is expected to be traded through
    quickly, not to act as a reaction zone."""
    for i in range(1, len(bars) - 1):
        bar = bars[i]
        body = abs(bar["close"] - bar["open"])
        full_range = bar["high"] - bar["low"] or 1e-9
        if body / full_range < 1 / body_ratio:
            continue  # not a thin/fast candle
        void_low, void_high = min(bar["open"], bar["close"]), max(bar["open"], bar["close"])
        last = bars[-1]
        if void_low <= last["low"] <= void_high:
            return Signal("Liquidity Void Fill", 2, "long", last["close"], void_low * 0.995, void_high)
    return None

Tier 3 — แนวคิด / เต็มด้วยดุลยพินิจ

Market Maker Buy Model (MMBM)

def mmbm(bars, window=15):
    """Tier 3. Full accumulation-manipulation-distribution cycle read — no
    single candle triggers this, only the phase sequence."""
    recent = bars[-window:]
    if len(recent) < window:
        return None
    accumulation = recent[: window // 3]
    acc_high = max(b["high"] for b in accumulation)
    acc_low = min(b["low"] for b in accumulation)
    manipulation = recent[window // 3 : 2 * window // 3]
    swept = any(b["low"] < acc_low for b in manipulation)
    distribution = recent[2 * window // 3 :]
    reclaimed = distribution[-1]["close"] > acc_high
    if swept and reclaimed:
        return Signal(
            "Market Maker Buy Model", 3, "long", distribution[-1]["close"], acc_low * 0.99,
            note="Phase-based read — confirm the phase boundaries yourself",
        )
    return None

Venom Model

def venom_model(bars):
    """Tier 3. A liquidity run plus an FVG entry, deliberately without the
    CHoCH confirmation the 2022 Model requires — looser by design."""
    low_swings = [s for s in find_swings(bars) if s.kind == "low"]
    if len(low_swings) < 2 or low_swings[-1].price >= low_swings[-2].price:
        return None
    gaps = find_fvg(bars)
    if not gaps:
        return None
    gap = gaps[-1]
    last = bars[-1]
    if gap["low"] <= last["low"] <= gap["high"]:
        return Signal(
            "Venom Model", 3, "long", last["close"], gap["low"] * 0.995,
            note="No CHoCH required — confirm manually before trusting this",
        )
    return None

Smart Money Reversal (SMR)

def smr(bars):
    """Tier 3. The umbrella case: a sweep plus ANY structure shift, no
    confluence required — deliberately the loosest possible trigger."""
    swings = find_swings(bars)
    pools = find_liquidity_pools(swings)
    labeled = label_structure(swings)
    shifts = [s for s, _, brk in labeled if brk in ("BOS", "CHoCH")]
    if not shifts or not pools:
        return None
    if detect_sweep(bars, pools[-1]) is not None:
        return Signal(
            "Smart Money Reversal", 3, "long", bars[-1]["close"], None,
            note="Entry not specified — this label only confirms sweep+shift happened",
        )
    return None

Power of Three Daily Bias

def power_of_three_daily_bias(bars, session_bars=24):
    """Tier 3. Sets a directional BIAS from the AMD cycle — never a trigger
    on its own. Feed the bias into a Tier 1/2 function for the actual entry."""
    session = bars[-session_bars:]
    if len(session) < session_bars:
        return None
    acc = session[: session_bars // 3]
    acc_high, acc_low = max(b["high"] for b in acc), min(b["low"] for b in acc)
    later = session[session_bars // 3 :]
    swept_low = any(b["low"] < acc_low for b in later[: session_bars // 3])
    reclaimed = later[-1]["close"] > acc_high
    bias = "bullish" if swept_low and reclaimed else ("bearish" if not swept_low else "unclear")
    return {"strategy": "Power of Three Daily Bias", "tier": 3, "bias": bias,
            "note": "Bias only — not a signal, feed this into an entry strategy"}

Weekly Profile

def weekly_profile(bars_by_week):
    """Tier 3. Compares the current week's high/low against the prior
    week's. `bars_by_week`: a list of bar-lists, one per week, oldest first."""
    if len(bars_by_week) < 2:
        return None
    prior_week, this_week = bars_by_week[-2], bars_by_week[-1]
    prior_high = max(b["high"] for b in prior_week)
    prior_low = min(b["low"] for b in prior_week)
    swept_low = any(b["low"] < prior_low for b in this_week)
    broke_high = any(b["high"] > prior_high for b in this_week)
    bias = "bullish" if swept_low and broke_high else ("bearish" if broke_high and not swept_low else "mixed")
    return {"strategy": "Weekly Profile", "tier": 3, "bias": bias,
            "prior_high": prior_high, "prior_low": prior_low}

Quarterly Theory Reversal

def quarterly_theory_reversal(bars):
    """Tier 3. Splits the given range into 4 equal segments and flags a
    reversal candle near the Q4 boundary. Where "Q1" starts is NOT
    standardized across sources — you must supply a consistent convention."""
    n = len(bars)
    q_size = n // 4
    if q_size == 0:
        return None
    q4 = bars[3 * q_size :]
    if len(q4) < 2:
        return None
    is_bigger_reversal = (q4[-1]["close"] - q4[-1]["open"]) > (q4[0]["high"] - q4[0]["low"])
    if q4[-1]["close"] > q4[-1]["open"] and is_bigger_reversal:
        return Signal(
            "Quarterly Theory Reversal", 3, "long", q4[-1]["close"], q4[0]["low"] * 0.99,
            note="Q1 start convention is not standardized across sources",
        )
    return None

ดัชนีฟังก์ชัน

ดาวน์โหลดไฟล์ฉบับสมบูรณ์ (องค์ประกอบพื้นฐาน + ฟังก์ชันทั้ง 18 ฟังก์ชัน, ตรวจสอบไวยากรณ์และทดสอบเบื้องต้นกับข้อมูลสังเคราะห์ — ไม่พบข้อผิดพลาดขณะรัน แต่ "รันโดยไม่เกิดข้อผิดพลาด" ไม่ได้หมายความว่า "ทำกำไรได้"): smc_ict_strategies.py