麻将胡了

This commit is contained in:
2026-09-22 15:39:53 +08:00
parent 8f8d2e042c
commit 470d1471e5
70 changed files with 6497 additions and 150 deletions
+7
View File
@@ -0,0 +1,7 @@
"""麻将胡了 · 生成器包(规则参数 / 模型 / 引擎)。"""
from .config import * # noqa: F401,F403
from .engine import ( # noqa: F401
assert_no_native_wild, assert_wild_only_middle, count_scatters, new_grid,
play_round, play_spin, pos, settle_rotation, ways_wins,
)
from .models import Round, Rotation, Spin, Win # noqa: F401
+166
View File
@@ -0,0 +1,166 @@
"""麻将胡了 · 唯一事实来源(规则参数)。
数值全部取自 docs/麻将胡了设计方案.md(§2 规则 / §3 桶与 RTP / §2.8 投注档)。
改规则改这里;生成器、引擎、单测、仿真都从这里读。
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, Mapping, Tuple
# ============================================================
# 盘面:5 轴 × 4 行 = 20 格,ways = 4^5 = 1024
# ============================================================
GRID_COLS = 5
GRID_ROWS = 4
GRID_SIZE = GRID_ROWS * GRID_COLS
WAYS = GRID_ROWS ** GRID_COLS
WILD = "W" # 百搭(元宝):只能由金色符号转化产生,卷轴永不落
SCATTER = "S" # 胡(夺宝):只触发免费旋转,不赔现金
# 格位约定:接口/落库用 1 基、行优先(1~20);引擎内部用 0 基。
# 引擎内 index = row * GRID_COLS + col;pos() 负责换算(见 engine.py)。
@dataclass(frozen=True)
class SymbolRule:
code: str
name: str
payouts: Mapping[int, float] # 连轴数(3/4/5) → 倍投注总额
# 赔付表:原版"每条基础投注点数"已 ÷20 内化,表中数值**直接是倍率**
SYMBOL_RULES: Dict[str, SymbolRule] = {
"A": SymbolRule("A", "發", {3: 0.75, 4: 3.0, 5: 5.0}),
"B": SymbolRule("B", "中", {3: 0.5, 4: 2.0, 5: 4.0}),
"C": SymbolRule("C", "白板", {3: 0.4, 4: 1.0, 5: 3.0}),
"D": SymbolRule("D", "八萬", {3: 0.3, 4: 0.75, 5: 2.0}),
"E": SymbolRule("E", "花筒", {3: 0.2, 4: 0.5, 5: 1.0}),
"F": SymbolRule("F", "索红绿", {3: 0.2, 4: 0.5, 5: 1.0}),
"G": SymbolRule("G", "筒", {3: 0.1, 4: 0.25, 5: 0.5}),
"H": SymbolRule("H", "索绿", {3: 0.1, 4: 0.25, 5: 0.5}),
}
PAYING: Tuple[str, ...] = ("A", "B", "C", "D", "E", "F", "G", "H")
PAYTABLE: Dict[str, Mapping[int, float]] = {code: r.payouts for code, r in SYMBOL_RULES.items()}
MIN_CHAIN = 3 # 前缀链至少 3 连才成奖(只有 3/4/5 三档)
# 卷轴出现权重:只有 8 个付费符号 + 胡(**没有 W**,原生盘面永不落百搭)
SYMBOL_WEIGHTS: Dict[str, int] = {
"A": 5, "B": 6, "C": 8, "D": 9, "E": 11, "F": 11, "G": 13, "H": 13, "S": 2,
}
# 金色麻将符号:只可能出现在第 2/3/4 轴(0 基列 1..3)
GOLDEN_COLS: Tuple[int, ...] = (1, 2, 3)
GOLDEN_CELL_PROB = 0.10 # 逐格独立;新补入的符号也可以带金
# ============================================================
# 乘数阶梯(本波阶梯倍率):每旋从第一档重新开始,不跨旋累计
# ============================================================
LADDER_BASE: Tuple[int, ...] = (1, 2, 3, 5)
LADDER_FS: Tuple[int, ...] = (2, 4, 6, 10)
def ladder_mul(rotation_seq: int, is_fs: bool) -> int:
"""第 n 波(1 基)的阶梯倍率;第 4 波及以后取最后一档。"""
if rotation_seq < 1:
raise ValueError("波序从 1 开始")
ladder = LADDER_FS if is_fs else LADDER_BASE
return ladder[min(rotation_seq, len(ladder)) - 1]
# ============================================================
# 免费旋转:3+ 胡 → 12 旋 + 每多 1 个 +2;一屏最多 5 个胡;上限 36 旋
# "发不满就不发":已发放 + 本次应赠 > 36 时,本旋绝不落 3+ 胡
# ============================================================
SCATTER_TRIGGER = 3
FS_BASE_SPINS = 12
FS_EXTRA_PER_SCATTER = 2
MAX_SCATTER_PER_SCREEN = 5
MAX_FREE_SPINS = 36
def fs_award(scatter_count: int) -> int:
return FS_BASE_SPINS + FS_EXTRA_PER_SCATTER * max(0, scatter_count - SCATTER_TRIGGER)
# 单局封顶:总倍率到达即截断,剩余旋作废
MAX_WIN_CAP_MUL = 500
# 悬念盘:基础旋恰好 2 个胡且第 5 轴无胡(生成端配额,客户端播第 5 列放大)
NEARMISS_SCATTER_COUNT = 2
NEARMISS_2SCATTER_RATE = 0.015
# ============================================================
# 投注档:11 档纯投注档,无投注倍数(betMul 仅预留、服务端只接受 1)
# price = cost = box.bet;boxId = 序号(1~11)
# ============================================================
BET_LADDER: Tuple[int, ...] = (1, 3, 5, 10, 20, 30, 50, 100, 200, 300, 500)
BOX_IDS: Tuple[int, ...] = tuple(range(1, len(BET_LADDER) + 1))
def bet_of(box_id: int) -> int:
return BET_LADDER[box_id - 1]
# ============================================================
# 饰品桶(16 档,0x~500x):沿用玫瑰女郎 17 档结构、删掉 1000x 那档
# 期望倍率 = (区间下限 + 上限) / 2;区间连续无缝
# 桶 16 区间上界故意开到 700 作缓冲,但生成端"先按 500 截断再落桶"
# ============================================================
@dataclass(frozen=True)
class Bucket:
id: int
name: str
expect_mul: float
range_low: float
range_high: float
BUCKETS: Tuple[Bucket, ...] = (
Bucket(1, "0倍饰品", 0.0, 0.0, 0.0),
Bucket(2, "0.5倍饰品", 0.5, 0.01, 1.0),
Bucket(3, "1.5倍饰品", 1.5, 1.0, 2.0),
Bucket(4, "2.5倍饰品", 2.5, 2.0, 3.0),
Bucket(5, "4倍饰品", 4.0, 3.0, 5.0),
Bucket(6, "6.5倍饰品", 6.5, 5.0, 8.0),
Bucket(7, "10倍饰品", 10.0, 8.0, 12.0),
Bucket(8, "16倍饰品", 16.0, 12.0, 20.0),
Bucket(9, "25倍饰品", 25.0, 20.0, 30.0),
Bucket(10, "40倍饰品", 40.0, 30.0, 50.0),
Bucket(11, "60倍饰品", 60.0, 50.0, 70.0),
Bucket(12, "100倍饰品", 100.0, 70.0, 130.0),
Bucket(13, "150倍饰品", 150.0, 130.0, 170.0),
Bucket(14, "200倍饰品", 200.0, 170.0, 230.0),
Bucket(15, "300倍饰品", 300.0, 230.0, 370.0),
Bucket(16, "500倍饰品(封顶)", 500.0, 370.0, 700.0),
)
FIXED_CAP_BUCKET_ID = 16 # 封顶桶
def bucket_of(mul: float) -> int:
"""按**已截断**的倍率落桶(顺序:先 min(mul, 500) 再落桶)。<0.01 视为空转。"""
mul = min(float(mul), MAX_WIN_CAP_MUL)
if mul < 0.01:
return 1
for b in BUCKETS:
if b.range_low <= mul <= b.range_high:
return b.id
return FIXED_CAP_BUCKET_ID
# ============================================================
# RTP 与身份
# ============================================================
RTP = 0.80 # 目标返奖率(后台可按箱改)
CONVERGE_RATIO = 0.15 # 收敛阶段占比(一轮 N = fixedSum / (1 - cr))
AUDIENCE_COMMON = "common" # 其余全部(**含 user_type='03' 机器人**)
AUDIENCE_ANCHOR = "anchor" # user_type = '01'
AUDIENCE_WHITE = "white" # is_blank_account = 1(最高优先)
# 只有 3 个身份维度:robot 维度已按用户要求去掉(2026-09-22),机器人归 common
AUDIENCES: Tuple[str, ...] = (AUDIENCE_COMMON, AUDIENCE_ANCHOR, AUDIENCE_WHITE)
# 生成器/池子规模(先小规模验证,再放大到线上)
ALLOC_K = 1.5 # 池子 = 配比 × 1.5
ALLOC_FLOOR = 60 # 每桶保底条数
SMOKE_ROUNDS_PER_BUCKET = 20 # `--per-bucket 20` 那种冒烟规模
+279
View File
@@ -0,0 +1,279 @@
"""麻将胡了 · 规则引擎(服务端权威判定的唯一实现)。
与母本 olympus/engine.py 的差别(母本只记账、消除由 simulator 决定):
本引擎**从盘面自己判定**——1024 路前缀链、金色就地转百搭、列内重力+补位、阶梯、FS、500x 封顶。
生成端(mahjong_simulator.py)只负责"把盘面摆成想要的形状",摆完之后一切由这里算。
"""
from __future__ import annotations
import random
from dataclasses import replace
from typing import Dict, List, Optional, Sequence, Set, Tuple
from .config import (
GOLDEN_CELL_PROB, GOLDEN_COLS, GRID_COLS, GRID_ROWS, MAX_SCATTER_PER_SCREEN,
MAX_WIN_CAP_MUL, MIN_CHAIN, NEARMISS_SCATTER_COUNT, PAYING, PAYTABLE, SCATTER,
SCATTER_TRIGGER, SYMBOL_WEIGHTS, WILD, fs_award, ladder_mul,
)
from .models import Round, Rotation, Spin, Win
Cell = Optional[str] # None = 空洞(结算中间态,必被补位填满)
Grid = List[Cell]
# ---------------------------------------------------------------- 格位换算
def pos(row: int, col: int) -> int:
"""0 基 (行,列) → 接口/落库用的 1 基格位 1~20(行优先)。"""
return row * GRID_COLS + col + 1
def index_of(p: int) -> int:
"""1 基格位 → 0 基内部下标。"""
return p - 1
def _rc(i: int) -> Tuple[int, int]:
return divmod(i, GRID_COLS)
# ---------------------------------------------------------------- 盘面构造
def _pick(rng: random.Random) -> str:
"""卷轴落一个符号:只从 8 付费符号 + 胡里出(永不落 W)。"""
codes = tuple(SYMBOL_WEIGHTS)
return rng.choices(codes, weights=[SYMBOL_WEIGHTS[c] for c in codes])[0]
def _mark_golden(rng: random.Random, col: int, sym: str) -> bool:
"""金色只在第 2/3/4 轴,且只打在普通符号上(百搭/胡不变金)。"""
return (col in GOLDEN_COLS and sym != SCATTER and sym != WILD
and rng.random() < GOLDEN_CELL_PROB)
def new_grid(rng: random.Random) -> Tuple[Grid, Set[int]]:
"""开一旋:整盘新发 + 第 2/3/4 轴的普通符号按概率打金色。返回 (盘面, 金色格下标集)。"""
grid: Grid = []
golden: Set[int] = set()
for i in range(GRID_ROWS * GRID_COLS):
row, col = _rc(i)
sym = _pick(rng)
grid.append(sym)
if _mark_golden(rng, col, sym):
golden.add(i)
return grid, golden
def assert_no_native_wild(grid: Sequence[Cell]) -> None:
"""铁律:原生盘面(未结算)+ 补位都不产出 W,百搭只能由金色转化产生。"""
for i, v in enumerate(grid):
if v == WILD:
raise AssertionError(f"格位 {pos(*_rc(i))} 出现了非转化来源的百搭")
def assert_wild_only_middle(grid: Sequence[Cell]) -> None:
"""铁律:金色只在第 2/3/4 轴 ⇒ 百搭只可能在第 2/3/4 轴(第 1 轴永远没有)。"""
for i, v in enumerate(grid):
if v == WILD and _rc(i)[1] not in GOLDEN_COLS:
raise AssertionError(f"格位 {pos(*_rc(i))} 的百搭落在第 1 或第 5 轴,不可能")
def count_scatters(grid: Sequence[Cell]) -> int:
return sum(1 for v in grid if v == SCATTER)
def scatters_in_col(grid: Sequence[Cell], col: int) -> int:
return sum(1 for r in range(GRID_ROWS) if grid[r * GRID_COLS + col] == SCATTER)
# ---------------------------------------------------------------- 1024 路判定
def _count_in_col(grid: Sequence[Cell], col: int, sym: str) -> int:
"""该轴「符号 X 或百搭」的格数(金色格按自然符号计入)。"""
n = 0
for r in range(GRID_ROWS):
v = grid[r * GRID_COLS + col]
if v == sym or v == WILD:
n += 1
return n
def chain_cells(grid: Sequence[Cell], sym: str, run: int) -> Tuple[int, ...]:
cells = []
for col in range(run):
for r in range(GRID_ROWS):
if grid[r * GRID_COLS + col] in (sym, WILD):
cells.append(pos(r, col))
return tuple(cells)
def ways_wins(grid: Sequence[Cell]) -> Tuple[Win, ...]:
"""前缀链:第 1 轴起 1→2→3→4→5 连续,一旦断链整笔作废;ways 只乘链内轴。"""
wins: List[Win] = []
for sym in PAYING:
run, ways, counts = 0, 1, []
for col in range(GRID_COLS):
n = _count_in_col(grid, col, sym)
if n == 0:
break
run += 1
ways *= n
counts.append(n)
if run >= MIN_CHAIN:
wins.append(Win(
symbol=sym, run=run, counts=tuple(counts), ways=ways,
loc=chain_cells(grid, sym, run), pay_base=PAYTABLE[sym][run] * ways,
))
return tuple(wins)
# ---------------------------------------------------------------- 重力 + 补位
def collapse_fill(grid: Grid, golden: Set[int], rng: random.Random
) -> Tuple[Tuple[Tuple[int, int], ...], Tuple[Tuple[int, str], ...]]:
"""列内重力:幸存符号(含刚转化的百搭)保持顺序压到底,顶部补新符号(可带金色)。
金色是**镀在符号上**的:幸存符号下落时,金色标记要跟着它一起走,
否则会留在原格"传染"给新补进来的牌(也就不再是"那个金色符号"了)。
"""
drops: List[Tuple[int, int]] = []
fills: List[Tuple[int, str]] = []
for col in range(GRID_COLS):
survivors = [(r, grid[r * GRID_COLS + col], (r * GRID_COLS + col) in golden)
for r in range(GRID_ROWS) if grid[r * GRID_COLS + col] is not None]
for r in range(GRID_ROWS):
golden.discard(r * GRID_COLS + col) # 清空本列标记,按落点重打
new_col: List[Cell] = [None] * GRID_ROWS
offset = GRID_ROWS - len(survivors)
for k, (r, v, was_golden) in enumerate(survivors):
dst_row = offset + k
new_col[dst_row] = v
if was_golden:
golden.add(dst_row * GRID_COLS + col)
if dst_row != r:
drops.append((pos(r, col), pos(dst_row, col)))
for r in range(offset):
sym = _pick(rng)
new_col[r] = sym
fills.append((pos(r, col), sym))
if _mark_golden(rng, col, sym):
golden.add(r * GRID_COLS + col)
for r in range(GRID_ROWS):
grid[r * GRID_COLS + col] = new_col[r]
return tuple(drops), tuple(fills)
# ---------------------------------------------------------------- 一波结算
def settle_rotation(grid: Grid, golden: Set[int], seq: int, is_fs: bool,
rng: random.Random) -> Optional[Rotation]:
"""结算一波;无成奖返回 None。会就地改写 grid / golden。"""
wins = ways_wins(grid)
if not wins:
return None
involved: Set[int] = {index_of(p) for w in wins for p in w.loc}
rotation_mul = ladder_mul(seq, is_fs)
base_mul = sum(w.pay_base for w in wins)
# ① 全部中奖格"被消除",但金色格是就地改写:不空洞、不补牌
converted: List[Tuple[int, str]] = []
for i in sorted(involved):
if i in golden:
golden.discard(i)
converted.append((i, grid[i] or ""))
grid[i] = WILD
converted_idx = {i for i, _ in converted}
for i in involved:
if i not in converted_idx:
grid[i] = None
# ② 重力下落 + 顶部补位(刚转化的 W 也是普通盘面符号,一起落)
drops, fills = collapse_fill(grid, golden, rng)
assert_wild_only_middle(grid)
return Rotation(
seq=seq, rotation_mul=rotation_mul, wins=wins, base_mul=base_mul,
win_mul=base_mul * rotation_mul,
converted=tuple((pos(*_rc(i)), s) for i, s in converted),
drops=drops, fills=fills, grid=tuple(grid),
)
# ---------------------------------------------------------------- 一旋 / 一局
def _strip_scatters(grid: Grid, golden: Set[int], rng: random.Random,
keep: int = SCATTER_TRIGGER - 1) -> None:
"""把落定的胡数压到 keep(用于"一屏最多 5 个胡"与"发不满就不发")。"""
hits = [i for i, v in enumerate(grid) if v == SCATTER]
for i in hits[keep:]:
row, col = _rc(i)
grid[i] = _pick(rng)
golden.discard(i)
if _mark_golden(rng, col, grid[i] or ""):
golden.add(i)
def play_spin(rng: random.Random, index: int, is_fs: bool,
grid: Optional[Grid] = None, golden: Optional[Set[int]] = None) -> Spin:
"""播一旋:落定(grid)→ 逐波结算 → 直到本波无成奖。"""
if grid is None or golden is None:
grid, golden = new_grid(rng)
_strip_scatters(grid, golden, rng, keep=MAX_SCATTER_PER_SCREEN)
scatters = count_scatters(grid) # 触发只看**落定盘面**(补位落下的胡不算)
rotations: List[Rotation] = []
seq = 1
while True:
rot = settle_rotation(grid, golden, seq, is_fs, rng)
if rot is None:
break
rotations.append(rot)
seq += 1
return Spin(index=index, is_fs=is_fs, rotations=tuple(rotations),
mul=sum(r.win_mul for r in rotations), scatters=scatters)
def _is_near_miss(grid: Sequence[Cell]) -> bool:
"""恰好 2 个胡、且第 5 轴没有胡(客户端在第 5 列播放大)。"""
return (count_scatters(grid) == NEARMISS_SCATTER_COUNT
and scatters_in_col(grid, GRID_COLS - 1) == 0)
def play_round(rng: random.Random, box_id: int = 1) -> Round:
"""完整一局:基础旋 →(3+ 胡)免费旋转 → 重触发 → 封顶截断。"""
round_ = Round(box_id=box_id)
total = 0.0
grid, golden = new_grid(rng)
_strip_scatters(grid, golden, rng, keep=MAX_SCATTER_PER_SCREEN) # 一屏最多 5 个胡
round_.near_miss = _is_near_miss(grid)
base = play_spin(rng, index=0, is_fs=False, grid=grid, golden=golden)
if base.scatters >= SCATTER_TRIGGER:
award = fs_award(base.scatters)
base = replace(base, fs_awarded=award)
round_.free_spins = award
round_.spins.append(base)
total += base.mul
played = 0
while played < round_.free_spins:
played += 1
grid, golden = new_grid(rng)
_strip_scatters(grid, golden, rng, keep=MAX_SCATTER_PER_SCREEN)
sc = count_scatters(grid)
retrigger = False
if sc >= SCATTER_TRIGGER:
extra = fs_award(sc)
if round_.free_spins + extra <= 36: # "发不满就不发"
round_.free_spins += extra
retrigger = True
else:
_strip_scatters(grid, golden, rng) # 本旋就是不落 3+ 胡
spin = play_spin(rng, index=played, is_fs=True, grid=grid, golden=golden)
if retrigger:
spin = replace(spin, retrigger=True)
round_.spins.append(spin)
total += spin.mul
if total >= MAX_WIN_CAP_MUL: # 触顶:剩余旋作废
round_.capped = True
break
round_.raw_mul = total
round_.actual_mul = min(total, float(MAX_WIN_CAP_MUL))
return round_
+58
View File
@@ -0,0 +1,58 @@
"""麻将胡了 · 数据模型(frozen dataclass,母本 olympus/models.py 同风格)。"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
@dataclass(frozen=True)
class Win:
"""一笔奖 = 某符号的一条前缀链。"""
symbol: str
run: int # 链长 3/4/5(从第 1 轴起连续的轴数)
counts: Tuple[int, ...] # 链内各轴的格数(百搭计入)
ways: int # counts 的乘积
loc: Tuple[int, ...] # 本笔涉及的全部格位(1 基,含金色格与百搭)
pay_base: float # 赔付表[符号][链长] × ways,单位 = 倍投注总额
@dataclass(frozen=True)
class Rotation:
"""一波 = 一次消除结算。"""
seq: int # 本旋内第几波(= 阶梯序号)
rotation_mul: int # 本波阶梯倍率
wins: Tuple[Win, ...]
base_mul: float # Σ pay_base
win_mul: float # base_mul × rotation_mul
converted: Tuple[Tuple[int, str], ...] # (格位1基, 原自然符号):金色就地转百搭
drops: Tuple[Tuple[int, int], ...] # (from1基, to1基)
fills: Tuple[Tuple[int, str], ...] # (格位1基, 新符号)
grid: Tuple[Optional[str], ...] # 本波结算+补位后的盘面(20 格短码)
@dataclass(frozen=True)
class Spin:
"""一旋。"""
index: int # 0 = 基础旋(触发盘),>0 = 第 N 次免费旋转
is_fs: bool
rotations: Tuple[Rotation, ...]
mul: float # 本旋倍率
scatters: int # 本旋**落定盘面**上的胡数(补位产生的不算触发)
fs_awarded: int = 0 # 本旋触发的赠送旋数(0 = 没触发)
retrigger: bool = False
@dataclass
class Round:
"""一局 = 基础旋 + 免费旋转(可重触发),受 500x 封顶。"""
box_id: int
spins: List[Spin] = field(default_factory=list)
raw_mul: float = 0.0 # 封顶前的总倍率
actual_mul: float = 0.0 # min(raw_mul, 500) —— 落桶/发奖用这个
capped: bool = False
near_miss: bool = False # 基础旋恰好 2 个胡且第 5 轴无胡
free_spins: int = 0 # 本局发放的免费旋数
@property
def has_fs(self) -> bool:
return self.free_spins > 0