406 lines
20 KiB
Python
406 lines
20 KiB
Python
"""麻将胡了 · 回合池生成器：把引擎跑出的回合写进 `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()
|