| |
| """d1 vs d1 blackjack battle. |
| |
| Two llama.cpp servers, each hosting a LiquidAI d1-3B decision model, play |
| blackjack against a rule-based dealer. Every decision (bet size + hit / stand / |
| double) is a /v1/systemone query. The complete battle is written as JSON and |
| rendered into a static web page by build_site.py. |
| |
| Usage: |
| python3 blackjack.py --rounds 12 --out battle.json |
| """ |
| import argparse |
| import json |
| import random |
| import time |
| import urllib.error |
| import urllib.request |
| from datetime import datetime, timezone |
| from pathlib import Path |
|
|
| SUITS = ["♠", "♥", "♦", "♣"] |
| RANKS = ["A", "2", "3", "4", "5", "6", "7", "8", "9", "10", "J", "Q", "K"] |
| RANK_VALUE = {r: (11 if r == "A" else 10 if r in ("10", "J", "Q", "K") else int(r)) for r in RANKS} |
|
|
| DECKS = 6 |
| BASE_BET = 10 |
| START_BANKROLL = 1000 |
| BET_FRACTIONS = {"min": 0.50, "standard": 0.70, "max": 1.00} |
|
|
| AGENTS = [ |
| { |
| "id": "a", |
| "name": "阿保", |
| "style": "保守算牌派", |
| "description": "以保住本金為第一優先，只有在牌面明顯有利時才加注，遇到風險寧可停牌。", |
| "url": "http://127.0.0.1:8081", |
| }, |
| { |
| "id": "b", |
| "name": "阿衝", |
| "style": "激進壓迫派", |
| "description": "相信進攻是最好的防守，喜歡加倍與大注，願意為了期望值承擔波動。", |
| "url": "http://127.0.0.1:8082", |
| }, |
| ] |
|
|
| RULES = { |
| "decks": DECKS, |
| "dealer": "莊家軟 17 停牌 (S17)", |
| "blackjack_pays": "3:2", |
| "double": "僅限硬 9–11 或軟 13–18 時加倍，加倍後只補一張", |
| "split": "本局不開放分牌", |
| "insurance": "不開放保險", |
| "reshuffle": "剩餘牌數低於 25% 時重洗", |
| "start_bankroll": START_BANKROLL, |
| "betting": "注碼依當前資金比例：小注 50%、大注 70%、全押 100%（取整到 10）", |
| "rebuy": "無補充籌碼，兩邊都歸零才結束", |
| } |
|
|
|
|
| def make_shoe(rng): |
| shoe = [(r, s) for _ in range(DECKS) for s in SUITS for r in RANKS] |
| rng.shuffle(shoe) |
| return shoe |
|
|
|
|
| def card_str(card): |
| return f"{card[0]}{card[1]}" |
|
|
|
|
| def hand_value(cards): |
| total = sum(RANK_VALUE[r] for r, _ in cards) |
| soft_aces = sum(1 for r, _ in cards if r == "A") |
| while total > 21 and soft_aces: |
| total -= 10 |
| soft_aces -= 1 |
| return total, soft_aces > 0 |
|
|
|
|
| def is_blackjack(cards): |
| return len(cards) == 2 and hand_value(cards)[0] == 21 |
|
|
|
|
| def hi_lo(card): |
| r = card[0] |
| if r in ("2", "3", "4", "5", "6"): |
| return 1 |
| if r in ("10", "J", "Q", "K", "A"): |
| return -1 |
| return 0 |
|
|
|
|
| def shoe_state(shoe, running_count): |
| remaining = len(shoe) |
| decks_left = max(remaining / 52.0, 0.15) |
| counts = {r: 0 for r in RANKS} |
| for r, _ in shoe: |
| counts[r] += 1 |
| return { |
| "decks": DECKS, |
| "remaining_cards": remaining, |
| "penetration_pct": round(100 * (1 - remaining / (DECKS * 52))), |
| "cards_left_by_rank": counts, |
| "hi_lo_running_count": running_count, |
| "true_count": round(running_count / decks_left, 2), |
| } |
|
|
|
|
| class D1Client: |
| def __init__(self, agent): |
| self.agent = agent |
| self.url = agent["url"].rstrip("/") |
| self.calls = 0 |
| self.total_ms = 0 |
|
|
| def ask(self, state, questions, retries=2): |
| payload = json.dumps({"state": state, "questions": questions}).encode() |
| last_err = None |
| for _ in range(retries + 1): |
| started = time.time() |
| req = urllib.request.Request( |
| f"{self.url}/v1/systemone", |
| data=payload, |
| headers={"Content-Type": "application/json"}, |
| ) |
| try: |
| with urllib.request.urlopen(req, timeout=120) as resp: |
| body = json.loads(resp.read()) |
| elapsed = int((time.time() - started) * 1000) |
| self.calls += 1 |
| self.total_ms += elapsed |
| return body["answers"], elapsed, None |
| except Exception as exc: |
| last_err = f"{type(exc).__name__}: {exc}" |
| time.sleep(0.4) |
| return None, 0, last_err |
|
|
|
|
| def pick(answers, key, valid): |
| ans = (answers or {}).get(key) or {} |
| choice = ans.get("choice") |
| if choice not in valid: |
| return None |
| return ans |
|
|
|
|
| def draw(shoe, count): |
| cards = [shoe.pop() for _ in range(min(count, len(shoe)))] |
| return cards |
|
|
|
|
| def fallback_action(total, soft, allowed): |
| if "double" in allowed and total in (10, 11): |
| return "double" |
| if soft and total >= 19: |
| return "stand" |
| if total >= 17: |
| return "stand" |
| return "hit" |
|
|
|
|
| def play_hand(client, shoe, player, hand, dealer_up, running_count, rng): |
| steps = [] |
| while True: |
| total, soft = hand_value(hand["cards"]) |
| if total >= 21: |
| break |
| allowed = ["hit", "stand"] |
| double_spot = (not soft and 9 <= total <= 11) or (soft and 13 <= total <= 18) |
| if len(hand["cards"]) == 2 and player["bankroll"] >= hand["bet"] and double_spot: |
| allowed.append("double") |
|
|
| state = { |
| "game": "blackjack", |
| "you": { |
| "name": client.agent["name"], |
| "style": client.agent["style"], |
| "personality": client.agent["description"], |
| "bankroll": player["bankroll"], |
| }, |
| "hand": { |
| "cards": [card_str(c) for c in hand["cards"]], |
| "total": total, |
| "soft": soft, |
| "is_pair": len(hand["cards"]) == 2 and hand["cards"][0][0] == hand["cards"][1][0], |
| }, |
| "dealer_upcard": card_str(dealer_up), |
| "bet": hand["bet"], |
| "allowed_actions": allowed, |
| "shoe": shoe_state(shoe, running_count), |
| "rules": RULES, |
| } |
| criteria = { |
| "hit": "再補一張牌：點數偏低、需要往 17 點以上推進時使用", |
| "stand": "停牌：點數已接近 21，或再補牌爆牌風險過高時使用", |
| } |
| if "double" in allowed: |
| criteria["double"] = "加倍下注：再多押一倍且只補一張，通常用在 9、10、11 點這種有把握的局面" |
| cards_text = " ".join(card_str(c) for c in hand["cards"]) |
| questions = { |
| "action": { |
| "type": "choice", |
| "instructions": ( |
| f"你是 {client.agent['name']}，風格「{client.agent['style']}」：" |
| f"{client.agent['description']} " |
| "你在玩 21 點，目標是讓手牌點數盡量接近 21 但不超過，並贏過莊家。" |
| f"你的手牌是 {cards_text}，合計 {total} 點" |
| f"（{'軟牌，A 可算 11 或 1' if soft else '硬牌'}）。" |
| f"莊家明牌是 {card_str(dealer_up)}。現在只能選一個動作，請選長期期望值最高的。" |
| ), |
| "criteria": criteria, |
| }, |
| "risk": { |
| "type": "score", |
| "instructions": "以你的風格判斷，現在這個局面對你有多不利？", |
| "criteria": ["非常有利", "稍微有利", "有點不利", "非常不利"], |
| }, |
| } |
| answers, latency, err = client.ask(state, questions) |
| decision = pick(answers, "action", allowed) |
| if decision is None: |
| if not allowed: |
| break |
| action = fallback_action(total, soft, allowed) |
| decision = { |
| "type": "choice", |
| "choice": action, |
| "probabilities": {a: (1.0 if a == action else 0.0) for a in allowed}, |
| "confidence": 0.0, |
| "fallback": True, |
| "error": err, |
| } |
| action = decision["choice"] |
| step = { |
| "type": "action", |
| "cards": [card_str(c) for c in hand["cards"]], |
| "total": total, |
| "soft": soft, |
| "dealer_up": card_str(dealer_up), |
| "allowed": allowed, |
| "decision": decision, |
| "risk": (answers or {}).get("risk"), |
| "latency_ms": latency, |
| "drawn": None, |
| "total_after": total, |
| } |
| steps.append(step) |
|
|
| if action == "stand": |
| break |
| if action == "double": |
| player["bankroll"] -= hand["bet"] |
| hand["bet"] *= 2 |
| hand["doubled"] = True |
| new_card = draw(shoe, 1)[0] |
| running_count += hi_lo(new_card) |
| step["drawn"] = card_str(new_card) |
| hand["cards"].append(new_card) |
| hand["total"], hand["soft"] = hand_value(hand["cards"]) |
| step["total_after"] = hand["total"] |
| break |
| new_card = draw(shoe, 1)[0] |
| running_count += hi_lo(new_card) |
| step["drawn"] = card_str(new_card) |
| hand["cards"].append(new_card) |
| hand["total"], hand["soft"] = hand_value(hand["cards"]) |
| step["total_after"] = hand["total"] |
| if hand["total"] > 21: |
| break |
|
|
| hand["total"], hand["soft"] = hand_value(hand["cards"]) |
| hand["bust"] = hand["total"] > 21 |
| hand["blackjack"] = is_blackjack(hand["cards"]) and not hand["doubled"] |
| return running_count, steps |
|
|
|
|
| def settle(hand, dealer_total, dealer_blackjack, dealer_bust): |
| if hand["bust"]: |
| return "lose", -hand["bet"] |
| if hand["blackjack"] and not dealer_blackjack: |
| return "blackjack", int(hand["bet"] * 1.5) |
| if dealer_blackjack and not hand["blackjack"]: |
| return "lose", -hand["bet"] |
| if dealer_bust: |
| return "win", hand["bet"] |
| if hand["total"] > dealer_total: |
| return "win", hand["bet"] |
| if hand["total"] < dealer_total: |
| return "lose", -hand["bet"] |
| return "push", 0 |
|
|
|
|
| def ask_bet(client, player, dealer_up_hint, shoe, running_count): |
| state = { |
| "game": "blackjack", |
| "you": { |
| "name": client.agent["name"], |
| "style": client.agent["style"], |
| "personality": client.agent["description"], |
| "bankroll": player["bankroll"], |
| }, |
| "phase": "betting", |
| "base_bet": BASE_BET, |
| "shoe": shoe_state(shoe, running_count), |
| "rules": RULES, |
| } |
| questions = { |
| "bet": { |
| "type": "choice", |
| "instructions": ( |
| f"你是 {client.agent['name']}，風格是「{client.agent['style']}」：" |
| f"{client.agent['description']} 新的一局要下注了，" |
| "請依資金、剩餘牌組成與真數決定注碼。" |
| ), |
| "criteria": { |
| "min": "小注：押上目前資金的一半", |
| "standard": "大注：押上目前資金的 70%", |
| "max": "全押：把目前資金全部押上", |
| }, |
| }, |
| "confidence": { |
| "type": "score", |
| "instructions": "你對這一注的信心有多高？", |
| "criteria": ["沒信心", "普通", "有信心", "非常有信心"], |
| }, |
| } |
| answers, latency, err = client.ask(state, questions) |
| decision = pick(answers, "bet", ["min", "standard", "max"]) |
| if decision is None: |
| decision = { |
| "choice": "min", |
| "probabilities": {"min": 1.0, "standard": 0.0, "max": 0.0}, |
| "confidence": 0.0, |
| "fallback": True, |
| "error": err, |
| } |
| bankroll = player["bankroll"] |
| fraction = BET_FRACTIONS.get(decision["choice"], 0.50) |
| amount = int(bankroll * fraction) |
| if amount >= BASE_BET: |
| amount = max(BASE_BET, int(round(amount / 10.0)) * 10) |
| bet = min(bankroll, max(amount, BASE_BET if bankroll >= BASE_BET else bankroll)) |
| if bet < BASE_BET: |
| bet = bankroll |
| if bet <= 0: |
| bet = bankroll |
| return { |
| "decision": decision, |
| "bet": max(bet, 0), |
| "confidence": (answers or {}).get("confidence"), |
| "latency_ms": latency, |
| } |
|
|
|
|
| def run(rounds, seed, out_path, until_broke=True, max_rounds=500, start_bankroll=START_BANKROLL): |
| rng = random.Random(seed) |
| shoe = make_shoe(rng) |
| running_count = 0 |
| players = [ |
| {"id": a["id"], "name": a["name"], "style": a["style"], "bankroll": start_bankroll} |
| for a in AGENTS |
| ] |
| peaks = {p["id"]: start_bankroll for p in players} |
| bust_round = {} |
| clients = {a["id"]: D1Client(a) for a in AGENTS} |
| started = time.time() |
| history = [] |
| limit = max_rounds if until_broke else rounds |
| n = 0 |
|
|
| while True: |
| if until_broke and all(p["bankroll"] <= 0 for p in players): |
| break |
| if n >= limit: |
| break |
| n += 1 |
|
|
| if len(shoe) < DECKS * 52 * 0.25: |
| shoe = make_shoe(rng) |
| running_count = 0 |
|
|
| active = [a for a in AGENTS if next(p for p in players if p["id"] == a["id"])["bankroll"] > 0] |
| events = [] |
| for a in AGENTS: |
| if a not in active: |
| player = next(p for p in players if p["id"] == a["id"]) |
| events.append(f"{player['name']} 資金為 0，本局無法下注") |
|
|
| bets = {} |
| for a in active: |
| player = next(p for p in players if p["id"] == a["id"]) |
| bets[a["id"]] = ask_bet(clients[a["id"]], player, None, shoe, running_count) |
|
|
| hands = {} |
| for a in active: |
| player = next(p for p in players if p["id"] == a["id"]) |
| hands[a["id"]] = { |
| "cards": draw(shoe, 2), |
| "bet": bets[a["id"]]["bet"], |
| "doubled": False, |
| } |
| running_count += sum(hi_lo(c) for c in hands[a["id"]]["cards"]) |
| player["bankroll"] -= hands[a["id"]]["bet"] |
|
|
| dealer_cards = draw(shoe, 2) |
| running_count += sum(hi_lo(c) for c in dealer_cards) |
| dealer_up, dealer_hole = dealer_cards[0], dealer_cards[1] |
| dealer_blackjack = is_blackjack(dealer_cards) |
| peek = RANK_VALUE[dealer_up[0]] in (10, 11) |
| if peek: |
| events.append("莊家明牌為 A 或 10 點，進行暗牌確認") |
|
|
| round_rec = { |
| "n": n, |
| "shoe_before": shoe_state(shoe, running_count), |
| "dealer_up": card_str(dealer_up), |
| "dealer_hole": card_str(dealer_hole), |
| "dealer_revealed": False, |
| "dealer": None, |
| "events": events, |
| "players": [], |
| } |
|
|
| if dealer_blackjack: |
| events.append("莊家直接拿到 Blackjack，本局立即結束") |
| round_rec["dealer_revealed"] = True |
| round_rec["dealer"] = { |
| "cards": [card_str(c) for c in dealer_cards], |
| "total": 21, |
| "soft": hand_value(dealer_cards)[1], |
| "blackjack": True, |
| "bust": False, |
| } |
| else: |
| for a in active: |
| hand = hands[a["id"]] |
| player = next(p for p in players if p["id"] == a["id"]) |
| running_count, steps = play_hand( |
| clients[a["id"]], shoe, player, hand, dealer_up, running_count, rng |
| ) |
| hand["steps"] = steps |
| hand["player_id"] = a["id"] |
|
|
| if not dealer_blackjack: |
| dealer_total, dealer_soft = hand_value(dealer_cards) |
| if any(not h["bust"] for h in hands.values()): |
| if peek: |
| events.append("確認結果：莊家沒有 Blackjack") |
| while dealer_total < 17: |
| card = draw(shoe, 1)[0] |
| running_count += hi_lo(card) |
| dealer_cards.append(card) |
| dealer_total, dealer_soft = hand_value(dealer_cards) |
| events.append(f"莊家補牌 {card_str(card)} → {dealer_total} 點") |
| events.append(f"莊家最終 {dealer_total} 點" + ("，爆牌" if dealer_total > 21 else "")) |
| else: |
| events.append("所有玩家都爆牌，莊家不補牌") |
| round_rec["dealer_revealed"] = True |
| round_rec["dealer"] = { |
| "cards": [card_str(c) for c in dealer_cards], |
| "total": dealer_total, |
| "soft": dealer_soft, |
| "blackjack": False, |
| "bust": dealer_total > 21, |
| } |
|
|
| dealer_total = round_rec["dealer"]["total"] |
| dealer_bust = round_rec["dealer"]["bust"] |
|
|
| for a in active: |
| player = next(p for p in players if p["id"] == a["id"]) |
| hand = hands[a["id"]] |
| hand["blackjack"] = is_blackjack(hand["cards"]) and not hand["doubled"] |
| hand["bust"] = hand_value(hand["cards"])[0] > 21 |
| result, net = settle(hand, dealer_total, dealer_blackjack, dealer_bust) |
| hand.update( |
| { |
| "cards": [card_str(c) for c in hand["cards"]], |
| "total": hand_value(hand["cards"])[0], |
| "soft": hand_value(hand["cards"])[1], |
| "blackjack": is_blackjack(hand["cards"]) and not hand["doubled"], |
| "bust": hand_value(hand["cards"])[0] > 21, |
| "result": result, |
| "net": net, |
| "bet_decision": bets[a["id"]], |
| } |
| ) |
| player["bankroll"] += hand["bet"] + net |
| if player["bankroll"] < 0: |
| player["bankroll"] = 0 |
| peaks[a["id"]] = max(peaks[a["id"]], player["bankroll"]) |
| hand["bankroll_after"] = player["bankroll"] |
| hand["id"] = a["id"] |
| hand["name"] = a["name"] |
| hand["style"] = a["style"] |
| hand["peak_bankroll"] = peaks[a["id"]] |
| if player["bankroll"] == 0 and a["id"] not in bust_round: |
| bust_round[a["id"]] = n |
| events.append(f"{player['name']} 資金歸零，破產出局") |
| round_rec["players"].append(hand) |
|
|
| round_rec["shoe_after"] = shoe_state(shoe, running_count) |
| history.append(round_rec) |
| print( |
| f"round {n:02d} | dealer {dealer_total} | " |
| + " | ".join( |
| f"{h['name']} {h['total']} {h['result']} net {h['net']:+d} bank {h['bankroll_after']}" |
| for h in round_rec["players"] |
| ), |
| flush=True, |
| ) |
|
|
| meta = { |
| "title": "d1 × d1 21 點：破產對決", |
| "subtitle": "兩個 LiquidAI d1-3B 決策模型，用 llama.cpp 一路打到雙方資金歸零", |
| "created_at": datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds"), |
| "seed": seed, |
| "model": "LiquidAI/d1-3B-GGUF", |
| "quant": "Q8_0", |
| "llama_build": "b11786 (0.7.0-dev)", |
| "rules": RULES, |
| "start_bankroll": start_bankroll, |
| "bet_fractions": BET_FRACTIONS, |
| "end_reason": "both_broke" if all(p["bankroll"] <= 0 for p in players) else "round_limit", |
| "agents": [ |
| { |
| **a, |
| "final_bankroll": next(p for p in players if p["id"] == a["id"])["bankroll"], |
| "peak_bankroll": peaks[a["id"]], |
| "bust_round": bust_round.get(a["id"]), |
| "rounds_played": sum(1 for r in history if any(x["id"] == a["id"] for x in r["players"])), |
| } |
| for a in AGENTS |
| ], |
| "rounds_played": len(history), |
| "duration_sec": round(time.time() - started, 2), |
| "inference": { |
| a["id"]: { |
| "calls": clients[a["id"]].calls, |
| "avg_latency_ms": round(clients[a["id"]].total_ms / max(clients[a["id"]].calls, 1)), |
| } |
| for a in AGENTS |
| }, |
| } |
| data = {"meta": meta, "rounds": history} |
| out = Path(out_path) |
| out.parent.mkdir(parents=True, exist_ok=True) |
| out.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8") |
| print(f"wrote {out} ({out.stat().st_size / 1024:.1f} KiB) | end: {meta['end_reason']} | rounds: {len(history)}") |
| return data |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--rounds", type=int, default=12) |
| parser.add_argument("--seed", type=int, default=20261021) |
| parser.add_argument("--out", default="battle.json") |
| parser.add_argument("--max-rounds", type=int, default=500) |
| parser.add_argument("--start-bankroll", type=int, default=START_BANKROLL) |
| parser.add_argument("--fixed-rounds", action="store_true", help="stop after --rounds instead of playing until both are broke") |
| args = parser.parse_args() |
| run(args.rounds, args.seed, args.out, until_broke=not args.fixed_rounds, |
| max_rounds=args.max_rounds, start_bankroll=args.start_bankroll) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
