source/blackjack.py

#!/usr/bin/env python3
"""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:  # noqa: BLE001
                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()