HKJC 即場賠率與海外博彩公司即時套利：從數學、資料到 Python 監控Script
- 4月26日
- 讀畢需時 9 分鐘
即場賠率的 arbitrage betting，表面上只是「在不同莊家之間買齊所有結果，令任何賽果都有正回報」。但真正落到工程層面，它其實是一個低延遲資料整合、事件匹配、風險控制與交易執行問題。尤其把 HKJC 的即場賠率與海外博彩公司報價並列時，挑戰不在公式，而在於盤口是否真的是同一市場、價格是否仍可成交、投注額是否受限，以及賠率在你按下確認前是否已改變。
本文用技術角度拆解這個問題：先由套利數學開始，再講即時 odds pipeline、資料標準化、市場匹配、延遲風險、stake sizing、外匯與手續費，最後給一個可跑的 Python 監控範例。範例只讀取你有權使用的 odds feed 或本機 JSON snapshot，不包含繞過網站條款、破解 API、規避地區限制或自動下注的內容。
重要聲明：博彩受地區法律、平台條款、年齡限制與負責任博彩規範約束。香港用戶尤其要留意本地法規以及海外投注平台是否合法提供服務。本文只作資訊安全、金融工程與資料工程角度的技術說明，不構成博彩建議、投資建議或鼓勵下注。
一、Arbitrage 的核心數學
十進制賠率（decimal odds）下，某結果賠率為 odds，其隱含概率為：
implied_probability = 1 / odds若一場足球賽的 1X2 市場有三個結果：主勝、和局、客勝。假設我們可以在不同公司買到各自最高賠率：
套利條件是所有結果的隱含概率總和小於 1：
1/2.20 + 1/3.60 + 1/4.40 = 0.4545 + 0.2778 + 0.2273 = 0.95960.9596 < 1，代表理論上存在約 1 - 0.9596 = 4.04% 的 gross arbitrage margin。若總投注本金為 B，每個結果的 stake 可按下式分配：
stake_i = (B / odds_i) / sum(1 / odds_j)這樣每個結果的 payout 會接近相同：
payout_i = stake_i * odds_i
profit_i = payout_i - B在理想世界，這是無風險套利。但在走地市場，「理想世界」幾乎不存在。你看到的價格、你計算的 margin、你可投注的額度、你提交後實際成交的 odds，中間都有時間差與操作風險。
二、為何即場套利比賽前套利難得多
賽前市場的價格更新相對慢，盤口定義較穩定；走地市場則完全不同。足球入球、紅牌、VAR、傷停、角球、危險進攻，都可能令 odds 在數秒內跳動。套利 scanner 在數學上找到機會，不代表交易上可以落地。
第一個難點是 latency。你讀到 HKJC snapshot 的時間可能是 t0，海外 bookmaker snapshot 是 t0 + 500ms，你完成計算是 t0 + 800ms，人工或程式準備下注已到 t0 + 2s。對走地盤來說，兩秒已經可以是另一個市場。
第二個難點是 market equivalence。同一場賽事的「主勝／和／客勝」看似一樣，但要檢查是否同一段時間、是否含加時、是否 90 分鐘市場、是否已扣除當前比分、是否 suspended 後重開、是否同一 handicap line。不同公司對市場命名與結算規則可能有差異，不能單靠隊名相似就當作同一市場。
第三個難點是 execution risk。套利通常要在至少兩個平台同時成交。若先成交一邊，另一邊 odds 改了、限額不足、投注被拒、帳戶需要額外確認，就會變成裸露風險。走地環境下，execution risk 往往比公式 margin 更重要。
三、HKJC 與海外 odds 的資料工程問題
一個可用的即時 arbitrage 系統，通常有五層：
Ingestion：從授權 API、付費 odds feed、內部手動輸入或合規 snapshot 取得賠率。
Normalization：統一 odds format、時間戳、隊名、賽事 ID、市場類型與 outcome label。
Matching：判斷不同來源是否對應同一 event 與同一 market。
Pricing Engine：找最佳賠率、計算 implied probability、margin、stake 與 sensitivity。
Risk Gate：檢查延遲、限額、手續費、外匯、最小投注、賠率變更與人工確認。
最容易被低估的是第二與第三層。HKJC 的球隊名稱、海外 bookmaker 的英文簡寫、第三方 feed 的 canonical team ID 可能全部不同。例如「曼聯」、「Manchester United」、「Man Utd」、「MUN」要映射到同一隊；「90 mins 1X2」與「To Win Match」則未必同一結算規則。
工程上建議建立一個 canonical schema：
{
"source": "hkjc",
"event_id": "canonical:event:2026-04-26:arsenal-chelsea",
"source_event_id": "HKJC-123456",
"sport": "football",
"league": "Premier League",
"start_time_utc": "2026-04-26T15:00:00Z",
"market": "match_winner_90m",
"in_play": true,
"score": "0-0",
"clock": "37:22",
"outcomes": [
{"name": "home", "odds": 2.2, "available": true},
{"name": "draw", "odds": 3.6, "available": true},
{"name": "away", "odds": 4.4, "available": true}
],
"observed_at": "2026-04-26T15:37:23.450Z"
}這個 schema 的重點不是欄位多，而是它把「來源資料」與「canonical interpretation」分開。所有套利計算都應只吃 canonical layer，避免在 pricing engine 裡到處寫 bookmaker-specific if/else。
四、套利機會不等於可交易機會
套利 scanner 常見誤報主要來自以下幾類。
Odds staleness：其中一邊賠率已經過期，但 feed 尚未更新。走地市場應為每個 quote 設定 max age，例如足球 1X2 走地不接受超過 2–5 秒的 quote；網球逐分市場可能更嚴。
Suspension mismatch：某平台因危險進攻或入球疑似事件暫停投注，另一平台仍顯示舊 odds。若系統未正確處理 available=false 或 suspended 狀態，會把不能交易的價格當成可用價格。
Line mismatch：讓球盤與大小球最容易出錯。例如 Over 2.5 與 Over 2.75 不是同一市場；亞洲盤還有半贏半輸結算，不能用簡單二元結果公式。
Commission and tax：交易所或部分平台可能有 commission；不同幣種入金、提款或換匯亦有成本。套利 margin 若只有 0.5%，很容易被成本吃掉。
Stake limits and rounding：理論 stake 可能是 HKD 173.42，但平台最小投注、最大投注、整數限制、單注上限會令 payout 不再完全平衡。
因此，專業系統不應只輸出「有 arb」，而要輸出 net margin after friction、最大可下注本金、最壞情境 profit、quote age 與每個 leg 的成交風險。
五、Python 範例：由 odds snapshot 找 1X2 套利
以下範例是一個完整但保守的 scanner。它不連接任何未授權網站，也不做自動下注；你可以把 HKJC 與海外 bookmaker 的授權 feed 轉成同一 JSON schema，再交給它計算。
假設目錄內有兩個檔案：
data/hkjc_live.json
data/bookmakers_live.json格式可以是多個 event snapshot：
[
{
"source": "hkjc",
"event_id": "canonical:event:arsenal-chelsea",
"market": "match_winner_90m",
"in_play": true,
"observed_at": "2026-04-26T15:37:23.450Z",
"outcomes": [
{"name": "home", "odds": 2.2, "available": true},
{"name": "draw", "odds": 3.55, "available": true},
{"name": "away", "odds": 4.05, "available": true}
]
}
]Scanner 程式
from __future__ import annotations
import json
import math
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable
OUTCOMES_1X2 = ("home", "draw", "away")
@dataclass(frozen=True)
class Quote:
source: str
event_id: str
market: str
outcome: str
odds: float
observed_at: datetime
available: bool = True
@property
def age_seconds(self) -> float:
return (datetime.now(timezone.utc) - self.observed_at).total_seconds()
@dataclass(frozen=True)
class ArbLeg:
outcome: str
source: str
odds: float
stake: float
quote_age_seconds: float
@dataclass(frozen=True)
class ArbOpportunity:
event_id: str
market: str
implied_sum: float
gross_margin: float
total_stake: float
equalized_payout: float
worst_case_profit: float
legs: tuple[ArbLeg, ...]
def parse_ts(value: str) -> datetime:
if value.endswith("Z"):
value = value[:-1] + "+00:00"
dt = datetime.fromisoformat(value)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
def load_quotes(path: Path) -> list[Quote]:
raw_events = json.loads(path.read_text(encoding="utf-8"))
quotes: list[Quote] = []
for event in raw_events:
observed_at = parse_ts(event["observed_at"])
for outcome in event["outcomes"]:
quotes.append(
Quote(
source=event["source"],
event_id=event["event_id"],
market=event["market"],
outcome=outcome["name"],
odds=float(outcome["odds"]),
observed_at=observed_at,
available=bool(outcome.get("available", True)),
)
)
return quotes
def best_quotes_by_outcome(
quotes: Iterable[Quote],
event_id: str,
market: str,
outcomes: tuple[str, ...],
max_quote_age_seconds: float,
) -> dict[str, Quote]:
best: dict[str, Quote] = {}
for quote in quotes:
if quote.event_id != event_id or quote.market != market:
continue
if quote.outcome not in outcomes:
continue
if not quote.available:
continue
if quote.odds <= 1.0 or not math.isfinite(quote.odds):
continue
if quote.age_seconds > max_quote_age_seconds:
continue
existing = best.get(quote.outcome)
if existing is None or quote.odds > existing.odds:
best[quote.outcome] = quote
return best
def compute_arb(
best: dict[str, Quote],
event_id: str,
market: str,
total_stake: float,
commission_rate: float = 0.0,
) -> ArbOpportunity | None:
if set(best) != set(OUTCOMES_1X2):
return None
# Simple commission model: reduce effective payout by a flat rate.
effective_odds = {
outcome: 1.0 + (quote.odds - 1.0) * (1.0 - commission_rate)
for outcome, quote in best.items()
}
implied_sum = sum(1.0 / odds for odds in effective_odds.values())
if implied_sum >= 1.0:
return None
equalized_payout = total_stake / implied_sum
legs: list[ArbLeg] = []
profits: list[float] = []
for outcome in OUTCOMES_1X2:
quote = best[outcome]
stake = equalized_payout / effective_odds[outcome]
payout = stake * effective_odds[outcome]
profits.append(payout - total_stake)
legs.append(
ArbLeg(
outcome=outcome,
source=quote.source,
odds=quote.odds,
stake=stake,
quote_age_seconds=quote.age_seconds,
)
)
return ArbOpportunity(
event_id=event_id,
market=market,
implied_sum=implied_sum,
gross_margin=1.0 - implied_sum,
total_stake=total_stake,
equalized_payout=equalized_payout,
worst_case_profit=min(profits),
legs=tuple(legs),
)
def scan_files(paths: list[Path], total_stake: float = 1000.0) -> list[ArbOpportunity]:
quotes: list[Quote] = []
for path in paths:
quotes.extend(load_quotes(path))
keys = sorted({(q.event_id, q.market) for q in quotes})
opportunities: list[ArbOpportunity] = []
for event_id, market in keys:
best = best_quotes_by_outcome(
quotes=quotes,
event_id=event_id,
market=market,
outcomes=OUTCOMES_1X2,
max_quote_age_seconds=5.0,
)
opportunity = compute_arb(
best=best,
event_id=event_id,
market=market,
total_stake=total_stake,
commission_rate=0.0,
)
if opportunity is not None:
opportunities.append(opportunity)
return opportunities
def main() -> None:
paths = [Path("data/hkjc_live.json"), Path("data/bookmakers_live.json")]
opportunities = scan_files(paths, total_stake=1000.0)
if not opportunities:
print("No arbitrage opportunity found.")
return
for arb in opportunities:
print(f"\n{arb.event_id} | {arb.market}")
print(f"implied_sum={arb.implied_sum:.4f}")
print(f"gross_margin={arb.gross_margin:.2%}")
print(f"equalized_payout={arb.equalized_payout:.2f}")
print(f"worst_case_profit={arb.worst_case_profit:.2f}")
for leg in arb.legs:
print(
f" {leg.outcome:>4} | {leg.source:<12} "
f"odds={leg.odds:.3f} stake={leg.stake:.2f} "
f"age={leg.quote_age_seconds:.2f}s"
)
if __name__ == "__main__":
main()這段程式的重點有三個。
第一，它用 event_id + market + outcome 作為計算 key，而不是直接拿「隊名字串」做 join。真實環境下，你應在前置流程把 HKJC 與海外 bookmaker 的賽事映射到 canonical event ID。
第二，它設定 max_quote_age_seconds=5.0。走地 odds 若超過可接受 age，即使數學上出現套利，也不應被視為可交易價格。不同運動與市場要用不同閾值，足球 1X2 與網球 point-by-point 市場不能同一標準。
第三，它把 commission 放入 effective odds。若某來源收取 2% commission，不能在找到套利後才扣，因為扣除後 implied sum 可能已經不小於 1。
六、加入外匯、限額與 rounding
真實系統通常要把每個來源的 stake 轉成該平台幣種，再處理投注單位。例如總本金以 HKD 計，海外 bookmaker 使用 GBP 或 USD，便要加入 FX rate、spread 與轉帳成本。
簡化公式如下：
stake_platform_currency = stake_hkd / fx_rate_hkd_per_unit
stake_rounded = round_to_platform_increment(stake_platform_currency)rounding 後要重新計算每個結果的 payout。不要假設四捨五入影響很小；當 arbitrage margin 低於 1% 時，最小投注單位與限額已足以把正收益變成負收益。
你可以把 ArbLeg 擴充為：
@dataclass(frozen=True)
class ExecutionConstraint:
source: str
currency: str
min_stake: float
max_stake: float
stake_increment: float
fx_rate_hkd_per_unit: float
commission_rate: float再在 compute_arb 後加一個 apply_execution_constraints()。這層應輸出三個結果：
只有當最差利潤仍大於你設定的 safety buffer，例如 max(20 HKD, 0.5% total_stake)，才值得提示人工檢查。
七、即時架構：Polling、WebSocket 與事件驅動
入門版本可以每 1–3 秒 polling 一次 odds snapshot，但專業即時系統通常會改成事件驅動：
Authorized Odds Feed
|
v
Ingestion Workers ----> Raw Quote Log
|
v
Normalizer ----> Canonical Quote Stream
|
v
Market Matcher ----> Matched Event Store
|
v
Pricing Engine ----> Opportunity Stream
|
v
Risk Gate ----> Alert / Manual ReviewRaw quote log 很重要，因為它讓你事後重播市場，檢查某個套利 alert 是否真的存在、是否由 stale quote 造成、是否 market mapping 出錯。若沒有 raw log，系統出現誤報時很難 debug。
技術選型方面，小型系統可用 asyncio + httpx + SQLite 或 Postgres；較高頻環境可用 Kafka／Redpanda 承接 quote stream，再由 pricing service 訂閱更新。重點不是追求複雜，而是每個 quote 必須保留 source_observed_at、ingested_at、normalized_at，讓你量度 latency budget。
八、風險控制：比公式更重要
一個成熟 scanner 應至少有以下風險 gate：
特別是 score consistency。走地 1X2 在 0-0 與 1-0 下完全不是同一市場狀態；若某來源比分更新慢，套利 scanner 可能看到極漂亮的 margin，但那其實是資料不同步。
九、回測：用歷史 odds 檢查 scanner 品質
若你能合法取得歷史 odds stream，應先做 backtesting，而不是直接接 alert。回測不是為了證明「一定賺」，而是量度三件事：
Signal frequency：每小時有多少個機會，分佈在哪些聯賽與市場。
False positive rate：扣除 stale quote、suspension mismatch、line mismatch 後剩多少。
Execution sensitivity：假設 1 秒、2 秒、5 秒延遲後，margin 還剩多少。
一個簡單做法是把歷史 quote 按時間排序，模擬每個時間點 scanner 看到的最新 quote，然後加入 execution delay：
def apply_execution_delay(quotes, delay_seconds):
# Conceptual example: at decision time t, assume execution happens at t + delay.
# Reprice each selected leg using the quote available at execution time.
...若機會在 2 秒延遲後大部分消失，這個策略就不適合人工執行；若要自動執行，又會牽涉平台條款、API 權限、風控與法律問題，不能只看技術可行性。
十、實務建議：把套利系統當成 risk system
走地套利最常見的錯誤，是把它當成純粹「找價差」問題。其實更合理的設計，是把它當成一個 risk system：scanner 只是提出候選，risk gate 決定是否值得人工檢查，execution layer 必須承認成交失敗與價格改變是常態。
對 HKJC 與海外 bookmaker 的組合而言，最低限度應做到：
只使用你有權存取與處理的 odds data。
明確標記每個 quote 的來源時間、接收時間與可交易狀態。
嚴格區分 90 分鐘、全場含加時、讓球、大小球等市場。
對每個 alert 顯示 quote age、net margin、最差利潤與 stake feasibility。
不把 0.2% 或 0.3% 的微小 margin 當成可執行機會，因為真實摩擦通常更大。
保留 raw quote log，方便事後審計與改善 mapping。
如果要做成產品級工具，下一步不是自動下注，而是建立一個可靠 dashboard：顯示 event mapping、每個來源最新 odds、latency、suspended 狀態、計算出的 stake，以及「為何某個機會被 risk gate 擋下」。這樣系統才可被驗證、可被審計，也不會因為漂亮的理論公式而忽略真實市場風險。
