"""麻将胡了 · 回合池生成器：把引擎跑出的回合写进 `game_mahjong_round` / `game_mahjong_step`。 两种模式： --mode natural （默认）自然形态：不干预，直接采样落库。**RTP≈5.25，只能测链路，不能当验收**。 --mode quota 配额构造：先标定每桶池内实测均值 → 二分解出 16 桶配额（空转率锁 65%、目标 RTP 0.80） → 按配额定量填充。**这才是能上线/能做 RTP 验收的池子**。 用法（在 mahjong-py 目录下）： python generate_rounds.py --mode quota --per-box 1000 --dry-run python generate_rounds.py --mode quota --per-box 1000 --write python generate_rounds.py --mode natural --per-box 100 --dry-run python generate_rounds.py --boxes 1,2 --per-box 100 --write --mode quota 写库内容（两种模式一样）： 1. `game_mahjong_round`：每局一行（bucket_id / actual_mul / 旋数 / 波数 / has_fs / near_miss） 2. `game_mahjong_step`：每帧一行（逐波；含 grid_state / ways_detail / 下落补位 / 阶梯 / 触发信息） 3. `game_mahjong_box_bucket`：**配额就是 count**（3 个身份各一套，used=0） 4. `game_mahjong_box_template`：每箱每身份一条生效模板（= 配额快照，供抽干补桶） 5. `game_mahjong_box.total_rounds`：回写成本次生成的回合数 """ from __future__ import annotations import argparse import json import math import random import sys from collections import Counter from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from mahjong import config as C # noqa: E402 from mahjong.engine import GRID_COLS, GRID_ROWS, play_round # noqa: E402 DB = dict(host="127.0.0.1", port=23306, user="henan", password="iTSJSPPZM3LSGAPC", database="henan") EMPTY_SHARE = 0.65 # 空转率（桶 1 配额占比，见设计文档 §3.9）；未按箱分档时的统一值 # 按档调波动（用户 2026-09-22 定）：**所有箱子 RTP 都锁 0.80**，但波动不同 —— # 低档（1~20）：空转低 → 剩余配额必须更"平" → 小奖更多、手感热； # 高档（200~500）：空转高 → 剩余配额更"陡" → 中大奖更多、博性更强。 # 机制：`solve_quotas` 会把"剩余部分"的加权均值凑到 `0.80/(1-空转率)`，所以 # **只改空转率就能同时改形状**，不用再单独调 power。 VOLATILITY_LADDER = ( (20, 0.62, "低档：空转 62%，小奖更多（热）"), (100, 0.65, "中档：空转 65%（设计基准）"), (10 ** 9, 0.68, "高档：空转 68%，中大奖更多（陡）"), ) def empty_share_for(box_id: int) -> float: bet = C.bet_of(box_id) for upper, share, _ in VOLATILITY_LADDER: if bet <= upper: return share return EMPTY_SHARE TARGET_RTP = 0.80 CALIB_ROUNDS = 6000 # 每箱标定样本（用来量每个桶的**池内实测**均值） MAX_TRIES = 400_000 # 单个桶的采样上限（防低概率桶把时间吃光） # ---------------------------------------------------------------- 行构造 def grid_rows(flat) -> list: """扁平 20 格 → 4 行 × 5 列（接口用的 gridState 形状）。""" return [list(flat[r * GRID_COLS:(r + 1) * GRID_COLS]) for r in range(GRID_ROWS)] def build_steps(round_, round_id: int): """把一局摊成逐帧行。""" rows, step_seq = [], 0 spin_win = 0.0 for spin in round_.spins: spin_win_before = spin_win acc_sym = Counter() winners = [] rotations = spin.rotations frames = rotations or (None,) # 空旋也要发一帧（落定盘面），否则客户端无从渲染 for i, rot in enumerate(frames, start=1): step_seq += 1 if rot is None: grid_flat = spin.landing ways, removed, drops, fills, conv, gold = [], [], [], [], [], [] base_mul, rot_mul = 0.0, C.ladder_mul(1, spin.is_fs) else: grid_flat = rot.grid_before gold = [{"pos": p, "symbol": rot.grid_before[p - 1]} for p in rot.golden_before] conv = [{"pos": p, "fromSymbol": s} for p, s in rot.converted] ways = [{"symbol": w.symbol, "reelCount": w.run, "counts": list(w.counts), "loc": list(w.loc), "ways": w.ways, "payBase": w.pay_base, "rotationMul": rot.rotation_mul, "win": round(w.pay_base * rot.rotation_mul, 4)} for w in rot.wins] seen = set() removed = [p for w in rot.wins for p in w.loc if not (p in seen or seen.add(p))] drops = [{"from": a, "to": b, "symbol": s} for a, b, s in rot.drops] fills = [{"pos": p, "symbol": s} for p, s in rot.fills] base_mul, rot_mul = rot.base_mul, rot.rotation_mul for w in rot.wins: acc_sym[w.symbol] += len(w.loc) winners.append({"symbol": w.symbol, "count": w.run, "win": round(w.pay_base * rot.rotation_mul, 4)}) spin_win = spin_win_before + sum(r.win_mul for r in rotations[:i]) is_last = (i == len(frames)) scat = ({"scatterCount": spin.scatters, "freeSpins": spin.fs_awarded, "extraSpins": 0} if spin.index == 0 and i == 1 and spin.fs_awarded else None) retri = ({"scatterCount": spin.scatters, "extraSpins": C.fs_award(spin.scatters)} if spin.index > 0 and i == 1 and spin.retrigger else None) rows.append(dict( id=round_id * 100 + step_seq, round_id=round_id, reel_spin_id=spin.index, rotation_seq=i, grid_state=json.dumps(grid_rows(grid_flat), ensure_ascii=False) if grid_flat else None, golden_detail=json.dumps(gold, ensure_ascii=False) if gold else None, wild_converted=json.dumps(conv, ensure_ascii=False) if conv else None, ways_detail=json.dumps(ways, ensure_ascii=False) if ways else None, accumulated_symbols=json.dumps(dict(acc_sym), ensure_ascii=False), spin_winners=json.dumps(winners, ensure_ascii=False), removed_positions=json.dumps(removed), drop_data=json.dumps(drops, ensure_ascii=False) if drops else None, fill_data=json.dumps(fills, ensure_ascii=False) if fills else None, scatter_trigger=json.dumps(scat, ensure_ascii=False) if scat else None, retrigger_info=json.dumps(retri, ensure_ascii=False) if retri else None, base_score=round(base_mul, 4), win_amount=round(sum(r.win_mul for r in rotations[:i]), 4), rotation_mul=rot_mul, spin_win=round(spin_win, 4), win_capped=1 if (round_.capped and is_last and spin is round_.spins[-1]) else 0, bigwin_level=None, is_final=1 if is_last else 0, )) spin_win = spin_win_before + sum(r.win_mul for r in rotations) return rows def make_round(box_id: int, seq: int, rng: random.Random): r = play_round(rng, box_id=box_id) round_id = box_id * 1_000_000 + seq steps = build_steps(r, round_id=round_id) row = dict(round_id=round_id, box_id=box_id, bucket_id=C.bucket_of(r.actual_mul), actual_mul=round(r.actual_mul, 4), free_spin_count=r.free_spins, total_rotations=sum(len(s.rotations) for s in r.spins), has_fs=1 if r.has_fs else 0, near_miss=1 if r.near_miss else 0) return row, steps # ---------------------------------------------------------------- 配额构造 def calibrate(box_id: int, n: int, seed: int): """小样本标定：每个桶的**池内实测均值**（自然均值与名义 expect_mul 差很多，必须实测）。""" rng = random.Random(seed) tot, cnt = Counter(), Counter() for _ in range(n): r = play_round(rng, box_id=box_id) b = C.bucket_of(r.actual_mul) tot[b] += r.actual_mul cnt[b] += 1 means = {} for b in C.BUCKETS: means[b.id] = (tot[b.id] / cnt[b.id]) if cnt[b.id] else b.expect_mul return means, cnt def _quota_from_power(means, power: float, empty_share: float): """w_b = 30×0.6^((b-1)×power)（b≥2），桶 1 锁 empty_share；返回 16 个配额（下标 1..16）。""" ws = {b.id: 30.0 * (0.6 ** ((b.id - 1) * power)) for b in C.BUCKETS if b.id >= 2} s = sum(ws.values()) q = {b.id: 0.0 for b in C.BUCKETS} q[1] = empty_share for b_id, w in ws.items(): q[b_id] = (1.0 - empty_share) * w / s return q def solve_quotas(means, target_rtp=TARGET_RTP, empty_share=EMPTY_SHARE): """二分 power，使 Σ q_b × mean_b = target_rtp。power 越大越偏向低倍桶（期望越低）。""" def expected(power): q = _quota_from_power(means, power, empty_share) return sum(q[b.id] * means[b.id] for b in C.BUCKETS) lo, hi = 0.05, 8.0 # expected(power) 随 power **递减**：power 越大越偏向低倍桶 for _ in range(80): mid = (lo + hi) / 2 if expected(mid) > target_rtp: lo = mid # 期望偏高 → 还要更大的 power else: hi = mid q = _quota_from_power(means, (lo + hi) / 2, empty_share) return q, expected((lo + hi) / 2) def _fill(box_id: int, targets: dict, seed: int, max_tries=MAX_TRIES): """按桶目标条数填充：落进目标桶才收。返回 (pool, bucket_cnt, short)。""" rng = random.Random(seed) pool, tries, short = [], Counter(), [] for b in C.BUCKETS: got = 0 while got < targets[b.id] and tries[b.id] < max_tries: tries[b.id] += 1 row, steps = make_round(box_id, len(pool) + 1, rng) if row["bucket_id"] == b.id: pool.append((row, steps)) got += 1 if got < targets[b.id]: short.append(f"桶{b.id} {got}/{targets[b.id]}") for i, (row, steps) in enumerate(pool, start=1): rid = box_id * 1_000_000 + i row["round_id"] = rid for j, st in enumerate(steps, start=1): st["round_id"] = rid st["id"] = rid * 100 + j cnt = Counter(row["bucket_id"] for row, _ in pool) return pool, cnt, short def quota_targets(per_box: int, quotas: dict): targets = {b.id: int(round(per_box * quotas[b.id])) for b in C.BUCKETS} diff = per_box - sum(targets.values()) if diff: targets[1] += diff return targets def _sample_bucket(box_id: int, bucket_id: int, rng: random.Random, max_tries=MAX_TRIES): """采一条落进指定桶的回合（低概率桶会多试几次）。""" for _ in range(max_tries): row, steps = make_round(box_id, 0, rng) if row["bucket_id"] == bucket_id: return row, steps return None def _tune(box_id: int, pool: list, target_rtp: float, per_box: int, seed: int, verbose=True): """整数可执行的定向微调：配额算出来是小数，四舍五入会把「0.4 条」的尾部桶抹成 0 条， 于是它本该贡献的期望就丢了（实测池内均值会系统性偏低 5~8pt）。 做法：拿空转局（桶 1，倍率 0）去换中高桶的回合，每换一条按**真实倍率**结账， 直到池内均值落进目标 ±0.005 以内。空转率只会动几条（<0.5pt）。 """ rng = random.Random(seed + 31337) cur = sum(r["actual_mul"] for r, _ in pool) need = target_rtp * per_box - cur # 还差多少（单位：倍率总和） empties = [i for i, (r, _) in enumerate(pool) if r["bucket_id"] == 1] ordered = sorted((x for x in C.BUCKETS if x.id >= 2), key=lambda x: -x.expect_mul) swaps = 0 while need > 0.6 and empties and ordered: cand = next((b for b in ordered if b.expect_mul <= need), None) if cand is None: break got = None for _ in range(300): # 只收"不超过缺口"的回合，保证不会冲过头 s = _sample_bucket(box_id, cand.id, rng) if s and s[0]["actual_mul"] <= need: got = s break if got is None: # 这一档试不出来，退一档 ordered = [b for b in ordered if b.id != cand.id] continue row, steps = got pool[empties.pop()] = (row, steps) need -= row["actual_mul"] swaps += 1 rng = random.Random(rng.random()) # 让后续采样换流 if verbose and swaps: print(f" 微调：换入 {swaps} 条中高桶回合（空转率随之动 {swaps / per_box:.2%}）") cnt = Counter(r["bucket_id"] for r, _ in pool) rtp = sum(r["actual_mul"] for r, _ in pool) / max(1, len(pool)) if verbose: print(f" 微调后池内实测 RTP {rtp:.4f}（与目标差 {abs(rtp - target_rtp):.4f}）") return pool, cnt def generate_quota(box_id: int, per_box: int, seed: int, empty_share: float, target_rtp: float, verbose=True): """解配额 → 定条数 → 填充 → 整数微调 → 重排 id。 `empty_share`：桶 1 的配额占比。CLI 不传时按 `empty_share_for(box_id)` 取**按档波动**的值 （低档 62% / 中档 65% / 高档 68%），这样各箱形状不同、但 RTP 都锁在 `target_rtp`。 为什么填完还要微调：运行时按库存消耗（一个 cycle 每条回合正好播一次）⇒ `cycle RTP = 池内均值`， 而配额是连续解出来的小数，四舍五入会抹掉尾部桶（见 `_tune` 的注释）。 """ means, _ = calibrate(box_id, CALIB_ROUNDS, seed) quotas, exp_rtp = solve_quotas(means, target_rtp, empty_share) targets = quota_targets(per_box, quotas) if verbose: print(f" 标定：配额理论期望 {exp_rtp:.4f}（目标 {target_rtp}）；桶1 配额 {quotas[1]:.2%}；" f"桶条数(1/2/3/12/16) = {targets[1]}/{targets[2]}/{targets[3]}/{targets[12]}/{targets[16]}") pool, cnt, short = _fill(box_id, targets, seed + 999 + box_id * 7919) pool, cnt = _tune(box_id, pool, target_rtp, per_box, seed + box_id, verbose) for i, (row, steps) in enumerate(pool, start=1): # 重排 id（采样与微调都打乱了顺序） rid = box_id * 1_000_000 + i row["round_id"] = rid for j, st in enumerate(steps, start=1): st["round_id"] = rid st["id"] = rid * 100 + j return pool, quotas, means, exp_rtp, short # ---------------------------------------------------------------- 写库 def write(conn, pool_by_box, per_box, audiences=C.AUDIENCES): cur = conn.cursor() total_r = total_s = 0 for box, (pool, quotas, bucket_cnt) in pool_by_box.items(): rounds = [row for row, _ in pool] steps = [st for _, sts in pool for st in sts] lo, hi = box * 1_000_000 + 1, box * 1_000_000 + per_box cur.execute(f"DELETE FROM game_mahjong_step WHERE round_id BETWEEN {lo} AND {hi}") cur.execute(f"DELETE FROM game_mahjong_round WHERE box_id={box}") cur.executemany( "INSERT INTO game_mahjong_round (round_id,box_id,bucket_id,actual_mul,free_spin_count," "total_rotations,has_fs,near_miss) VALUES (%(round_id)s,%(box_id)s,%(bucket_id)s," "%(actual_mul)s,%(free_spin_count)s,%(total_rotations)s,%(has_fs)s,%(near_miss)s)", rounds) cols = ("id", "round_id", "reel_spin_id", "rotation_seq", "grid_state", "golden_detail", "wild_converted", "ways_detail", "accumulated_symbols", "spin_winners", "removed_positions", "drop_data", "fill_data", "scatter_trigger", "retrigger_info", "base_score", "win_amount", "rotation_mul", "spin_win", "win_capped", "bigwin_level", "is_final") cur.executemany( f"INSERT INTO game_mahjong_step ({','.join(cols)}) " f"VALUES ({','.join('%(' + c + ')s' for c in cols)})", steps) for aud in audiences: cur.execute("DELETE FROM game_mahjong_box_bucket WHERE box_id=%s AND audience=%s", (box, aud)) cur.executemany( "INSERT INTO game_mahjong_box_bucket (box_id,bucket_id,audience,count,used) " "VALUES (%s,%s,%s,%s,0)", [(box, b.id, aud, bucket_cnt.get(b.id, 0)) for b in C.BUCKETS]) items = json.dumps([{"bucketId": b.id, "count": bucket_cnt.get(b.id, 0)} for b in C.BUCKETS], ensure_ascii=False) cur.execute("DELETE FROM game_mahjong_box_template WHERE box_id=%s AND audience=%s", (box, aud)) cur.execute( "INSERT INTO game_mahjong_box_template (box_id,audience,name,total_rounds,items_json,is_active) " "VALUES (%s,%s,%s,%s,%s,1)", (box, aud, f"配额池-{per_box}局", per_box, items)) cur.execute("UPDATE game_mahjong_box SET total_rounds=%s WHERE box_id=%s", (per_box, box)) conn.commit() total_r += len(rounds) total_s += len(steps) rtp = sum(r["actual_mul"] for r in rounds) / max(1, len(rounds)) print(f" 箱 {box:2d}（赌注 {C.bet_of(box):>3}）：回合 {len(rounds):>5} 帧 {len(steps):>6} " f"**池内实测 RTP {rtp:.4f}** 空转 {bucket_cnt.get(1,0)/max(1,len(rounds)):.1%} " f"FS 局 {sum(r['has_fs'] for r in rounds)}") print(f"合计：回合 {total_r}，帧 {total_s}") return total_r, total_s def build_natural(box_id, per_box, seed): rng = random.Random(seed + box_id) pool, cnt = [], Counter() for i in range(1, per_box + 1): row, steps = make_round(box_id, i, rng) pool.append((row, steps)) cnt[row["bucket_id"]] += 1 return pool, cnt def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--mode", choices=("natural", "quota"), default="natural") ap.add_argument("--per-box", type=int, default=1000) ap.add_argument("--boxes", default="") ap.add_argument("--seed", type=int, default=20260922) ap.add_argument("--empty-share", type=float, default=None, help="空转率（桶 1 配额占比）；不传=按档取（低档 0.62 / 中档 0.65 / 高档 0.68）") ap.add_argument("--target-rtp", type=float, default=TARGET_RTP) ap.add_argument("--write", action="store_true") ap.add_argument("--dry-run", action="store_true") ap.add_argument("--host", default=DB["host"]); ap.add_argument("--port", type=int, default=DB["port"]) ap.add_argument("--user", default=DB["user"]); ap.add_argument("--password", default=DB["password"]) ap.add_argument("--database", default=DB["database"]) args = ap.parse_args() boxes = [int(x) for x in args.boxes.split(",") if x.strip()] or list(C.BOX_IDS) print(f"==== {args.mode} 模式：{len(boxes)} 箱 × {args.per_box} 局" f"{'' if args.write and not args.dry_run else '（干跑，不写库）'} ====") pool_by_box = {} for box in boxes: if args.mode == "natural": pool, cnt = build_natural(box, args.per_box, args.seed) quotas = None pool_by_box[box] = (pool, quotas, cnt) else: share = args.empty_share if args.empty_share is not None else empty_share_for(box) pool, quotas, means, exp_rtp, short = generate_quota( box, args.per_box, args.seed, share, args.target_rtp) cnt = Counter(row["bucket_id"] for row, _ in pool) if short: print(f" 未填满：{'；'.join(short)}") pool_by_box[box] = (pool, quotas, cnt) if not (args.write and not args.dry_run): rows = [r for r, _ in pool_by_box[box][0]] rtp = sum(r["actual_mul"] for r in rows) / max(1, len(rows)) print(f" 箱 {box:2d}（赌注 {C.bet_of(box):>3}）：回合 {len(rows):>5} " f"**池内实测 RTP {rtp:.4f}** 空转 {cnt.get(1,0)/max(1,len(rows)):.1%} " f"FS 局 {sum(r['has_fs'] for r in rows)}") if not (args.write and not args.dry_run): return import pymysql conn = pymysql.connect(host=args.host, port=args.port, user=args.user, password=args.password, database=args.database, charset="utf8mb4") write(conn, pool_by_box, args.per_box) conn.close() if __name__ == "__main__": main()