"""复刻 Java 抽回合链路，模拟每个箱子抽完一个 cycle 的实时 RTP。 复刻的 Java 逻辑（OlympusRoundPicker / OlympusGameService / OlympusDistributionService）： 1. 固定池 fixedSum = Σ 配比 count（按身份），N = round(fixedSum/(1-收敛比例))， fixedRounds = round(N×(1-收敛比例))；固定阶段 drawn < fixedRounds 2. 固定阶段：在"remaining>0 且 expect_mul×bet ≤ payoutCap"的桶里按 remaining 加权抽； 桶内取 40 条候选，优先取 actual_mul×bet ≤ payoutCap 的（取不到就退回候选第一条）；该桶 used+1 3. 收敛阶段：desiredNextMul = (N×目标倍率 − 已派彩) / (N − drawn)， 取期望最接近 desired 的 3 个桶（随机其一），桶内取 actual_mul 最接近 min(desired, capMul) 的一条 （优先 ≤ capMul），不消耗 used 4. 水位：payoutCap = max(0, 水位 + 本局成本) × rtp；水位 += 成本 − 派彩（白号不参与水位 → 无上限） 5. drawn ≥ N → cycle 结束：resetCycle + refillFromConfig（有生效模板则按模板重写 count，否则 used 归零） 用法: python simulate_cycle.py # 全部 30 箱 × common 身份，跑 3 个 cycle python simulate_cycle.py --cycles 5 --audience white python simulate_cycle.py --box 21 """ import argparse import random from collections import defaultdict import olympus_db as db from olympus.config import ORNAMENTS, MODES, payout_expect # 抽回合的 payoutCap 过滤用**名义** expect_mul（对齐 Java OlympusRoundPicker，偏保守）； # 统计里的"名义 RTP"用 payout_expect（对齐后台「目标RTP」列的封顶修正口径）。 _ORN_MUL = {o['id']: o['expect_mul'] for o in ORNAMENTS} _ORN_PAY = {o['id']: payout_expect(o['id']) for o in ORNAMENTS} CANDIDATE_LIMIT = 40 def load_boxes(box_id=None): with db.get_conn() as conn: cur = conn.cursor() sql = 'SELECT box_id, box_type, bet, rtp, rtp_converge_ratio, total_ornaments FROM game_olympus_box' if box_id: sql += ' WHERE box_id=%s' % int(box_id) cur.execute(sql + ' ORDER BY box_id') boxes = [dict(zip(('box_id', 'box_type', 'bet', 'rtp', 'cr', 'total'), r)) for r in cur.fetchall()] cur.execute('SELECT box_id, ornament_id, audience, count FROM game_olympus_box_ornament') cfg = defaultdict(dict) for b, o, a, c in cur.fetchall(): cfg[(int(b), a)][int(o)] = int(c) cur.execute('SELECT box_id, ornament_id, actual_mul FROM game_olympus_round') rounds = defaultdict(lambda: defaultdict(list)) for b, o, m in cur.fetchall(): rounds[int(b)][int(o)].append(float(m)) # 生效模板（抽完重填的依据） cur.execute('''SELECT box_id, audience, items_json FROM game_olympus_box_template WHERE is_active=1 ORDER BY id''') tpl = {} import json for b, a, items in cur.fetchall(): tpl[(int(b), a)] = {int(i['ornamentId']): int(i['count']) for i in json.loads(items) if int(i.get('count', 0)) > 0} for b in boxes: b['cm'] = float(MODES[b['box_type']].cost_multiplier) return boxes, cfg, rounds, tpl def _weighted(cands, weights): total = sum(weights) r = random.uniform(0, total) acc = 0.0 for c, w in zip(cands, weights): acc += w if r <= acc: return c return cands[-1] def simulate(box, cfg, rounds, tpl, audience, cycles=3, cr_override=None): cm, rtp = box['cm'], float(box['rtp']) cr = float(cr_override if cr_override is not None else (box['cr'] or 0.15)) fixed_sum = sum(cfg.values()) n_total = round(fixed_sum / (1.0 - cr)) fixed_rounds = round(n_total * (1.0 - cr)) target_mul = rtp * cm capped = (audience != 'white') # 白号 payoutCap=null stats = [] counts = dict(cfg) cumulative = {'drawn': 0, 'mul': 0.0} balance = 0.0 # 箱子水位：跨 cycle 持久（= Σ投注 − Σ返奖） for cyc in range(1, cycles + 1): used = defaultdict(int) drawn, mul_sum = 0, 0.0 rtp_fixed = None while drawn < n_total: cap_mul = max(0.0, balance + cm) * rtp if capped else float('inf') remaining = {o: max(0, c - used[o]) for o, c in counts.items()} if drawn < fixed_rounds and sum(remaining.values()) > 0: elig = [o for o, r in remaining.items() if r > 0 and _ORN_MUL[o] <= cap_mul] or \ [o for o, r in remaining.items() if r > 0] orn = _weighted(elig, [remaining[o] for o in elig]) pool = rounds[box['box_id']].get(orn) or [0.0] cands = random.sample(pool, min(CANDIDATE_LIMIT, len(pool))) mul = next((m for m in cands if m <= cap_mul), cands[0]) used[orn] += 1 else: desired = max(0.0, (n_total * target_mul - mul_sum) / max(1, n_total - drawn)) allowed = [o for o, c in counts.items() if _ORN_MUL[o] <= cap_mul] or list(counts) allowed.sort(key=lambda o: abs(_ORN_MUL[o] - desired)) orn = random.choice(allowed[:3]) pool = rounds[box['box_id']].get(orn) or [0.0] eff = min(desired, cap_mul) mul = sorted(pool, key=lambda m: (0 if m <= cap_mul else 1, abs(m - eff)))[0] drawn += 1 mul_sum += mul balance += cm - mul cumulative['drawn'] += 1 cumulative['mul'] += mul if drawn == fixed_rounds: rtp_fixed = mul_sum / drawn / cm # cycle 结束 → 按模板重填（无模板则 used 归零、count 不变） tpl_counts = tpl.get((box['box_id'], audience)) refilled = '模板' if tpl_counts else 'used归零' if tpl_counts: counts = {o: c for o, c in tpl_counts.items() if c > 0} stats.append({ 'cycle': cyc, 'rtp_fixed': rtp_fixed, 'rtp_end': mul_sum / drawn / cm, 'balance_end': balance, 'refill': refilled, }) return { 'box_id': box['box_id'], 'box_type': box['box_type'], 'audience': audience, 'rtp': rtp, 'cr': cr, 'n_total': n_total, 'fixed_rounds': fixed_rounds, 'fixed_sum': fixed_sum, 'nominal': sum(c * _ORN_PAY.get(o, 0) for o, c in cfg.items()) / fixed_sum / cm, 'cycles': stats, 'cum_rtp': cumulative['mul'] / cumulative['drawn'] / cm, } def main(): ap = argparse.ArgumentParser() ap.add_argument('--box', type=int) ap.add_argument('--audience', default='common') ap.add_argument('--cycles', type=int, default=3) ap.add_argument('--cr', type=float, default=None, help='覆盖收敛比例（测敏感度用，不写库）') ap.add_argument('--seed', type=int, default=2026) args = ap.parse_args() random.seed(args.seed) boxes, cfg, rounds, tpl = load_boxes(args.box) print('身份=%s cycle 数=%d seed=%d 收敛比例=%s' % ( args.audience, args.cycles, args.seed, ('箱配置' if args.cr is None else args.cr))) print('箱 模式 配比sum N 固定池 名义RTP 各cycle(固定池末→整轮末) 累计RTP') rows = [] for b in boxes: c = cfg.get((b['box_id'], args.audience), {}) if not c: continue r = simulate(b, c, rounds, tpl, args.audience, args.cycles, args.cr) rows.append(r) per = ' '.join('%.1f→%.1f' % (s['rtp_fixed'] * 100, s['rtp_end'] * 100) for s in r['cycles']) flag = '✓' if abs(r['cum_rtp'] - r['rtp']) < 0.03 else '✗' print('%2d %-7s %5d %4d %5d %6.1f%% %-34s %6.1f%% %s' % ( r['box_id'], r['box_type'], r['fixed_sum'], r['n_total'], r['fixed_rounds'], r['nominal'] * 100, per, r['cum_rtp'] * 100, flag)) bad = [r for r in rows if abs(r['cum_rtp'] - r['rtp']) >= 0.03] print('\n达标(±3个百分点): %d/%d' % (len(rows) - len(bad), len(rows))) if bad: print('未达标:', [(r['box_id'], round(r['cum_rtp'] * 100, 1), round(r['rtp'] * 100, 1)) for r in bad]) if rows: r = rows[0] print('\n样例 box%d 逐 cycle 明细（%s）:' % (r['box_id'], r['audience'])) for s in r['cycles']: print(' cycle%d: 固定池结束 %.1f%% 整轮结束 %.1f%% 抽完水位=%.0f 倍底注 重填=%s' % ( s['cycle'], s['rtp_fixed'] * 100, s['rtp_end'] * 100, s['balance_end'], s['refill'])) if __name__ == '__main__': main()