490 lines
17 KiB
Python
490 lines
17 KiB
Python
import json
|
||
import random
|
||
from typing import List, Dict, Tuple, Optional
|
||
|
||
from symbol_pay_table import get_symbol_pay
|
||
|
||
# 符号默认权重
|
||
default_symbol_weights = {
|
||
"A": 2, # 高赔率
|
||
"B": 5, # 中高赔率
|
||
"C": 10, # 中等赔率
|
||
"D": 15, # 中低赔率
|
||
"E": 20, # 低赔率
|
||
"F": 30, # 极低赔率
|
||
"G": 40, # 最低赔率
|
||
"S": 1, # Scatter
|
||
}
|
||
|
||
scotter_count_mapping = {"3": 10, "4": 12, "5": 15, "6": 20, "7": 30}
|
||
|
||
|
||
class SugarRush1000:
|
||
def __init__(
|
||
self,
|
||
balance: float = 1000.0,
|
||
bet: float = 1.0,
|
||
mock_grid=None,
|
||
weights=None,
|
||
scotter_counts_weights=None,
|
||
):
|
||
"""
|
||
初始化游戏
|
||
:param balance: 玩家余额
|
||
:param bet: 单次下注额
|
||
"""
|
||
self.balance = balance
|
||
self.bet = bet
|
||
self.rows = 7
|
||
self.cols = 7
|
||
|
||
# 游戏符号定义和权重
|
||
self.symbol_weights = weights if weights else default_symbol_weights
|
||
self.symbols = list(self.symbol_weights.keys())
|
||
self.weights = list(self.symbol_weights.values())
|
||
|
||
if scotter_counts_weights is None:
|
||
scotter_counts_weights = {3: 10, 4: 12, 5: 15, 6: 20, 7: 30}
|
||
|
||
self.scotter_counts = list(scotter_counts_weights.keys())
|
||
self.scotter_counts_weights = list(scotter_counts_weights.values())
|
||
|
||
# 游戏状态
|
||
self.grid = []
|
||
# 7x7 网格
|
||
self.multipliers = [[0 for _ in range(self.cols)] for _ in range(self.rows)]
|
||
self.free_spins = 0 # 剩余免费旋转次数
|
||
self.is_free_spin_status = False
|
||
self.max_win_cap = 25000 # 最大赢利倍数限制
|
||
self.is_buy_free = False
|
||
self.is_super_free = False # 是否为超级免费旋转模式
|
||
self.spin_total_win = (
|
||
0 # 本次旋转的总赢利。主要计算免费旋转过程中累积盈利是否超过最大额度
|
||
)
|
||
self.mock_grid = mock_grid # 用于测试的固定网格
|
||
self.drop_sequence = [] # 用于测试的掉落序列
|
||
|
||
def _get_next_drop_symbol(self, count=1, no_scatter: bool = False):
|
||
"""生成一列中的count个符号，要求一列中最多有一个scatter符号
|
||
no_scatter: 是否排除scatter符号。如果设置为True，则不生成scatter符号，如果设置为False，则则本次生成的符号中顶多有一个scatter符号
|
||
"""
|
||
|
||
# 测试辅助方法：从预设序列中获取掉落符号。预设序列不使用最多有一个scatter符号的规则
|
||
result = []
|
||
while count > 0:
|
||
if self.drop_sequence:
|
||
result.append(self.drop_sequence.pop(0))
|
||
count -= 1
|
||
else:
|
||
break
|
||
|
||
symbols1 = self.symbols[:]
|
||
weights1 = self.weights[:]
|
||
tmp = [(s, w) for s, w in zip(symbols1, weights1) if s != "S"]
|
||
symbols2, weights2 = map(list, zip(*tmp))
|
||
for _ in range(count):
|
||
tmp_symbols = symbols1
|
||
tmp_weights = weights1
|
||
if no_scatter:
|
||
tmp_symbols = symbols2
|
||
tmp_weights = weights2
|
||
symbol = random.choices(tmp_symbols, weights=tmp_weights, k=1)[0]
|
||
result.append(symbol)
|
||
if symbol == "S":
|
||
no_scatter = True
|
||
return result
|
||
|
||
def grid_col_has_scotter(self, col):
|
||
for row in range(self.rows):
|
||
if self.grid[row][col] == "S":
|
||
return True
|
||
return False
|
||
|
||
def _init_grid(self):
|
||
"""初始化或重置网格"""
|
||
self.grid = [[None for _ in range(self.cols)] for _ in range(self.rows)]
|
||
for col in range(self.cols):
|
||
symbols = self._get_next_drop_symbol(count=self.rows, no_scatter=False)
|
||
for row in range(self.rows):
|
||
self.grid[row][col] = symbols[row]
|
||
|
||
def buy_free_spins(self, buy_type: Optional[str] = None) -> int:
|
||
""":param buy_type: None=普通旋转, 'standard'=买免费旋转(50x), 'super'=买超级免费旋转(200x)"""
|
||
result = {}
|
||
if not buy_type:
|
||
result = {"message": "未购买任何功能", "balance": self.balance}
|
||
|
||
cost = 0
|
||
if buy_type == "standard":
|
||
cost = self.bet * 50
|
||
elif buy_type == "super":
|
||
cost = self.bet * 200
|
||
|
||
if self.balance < cost:
|
||
result["message"] = "余额不足，无法购买功能"
|
||
return result
|
||
|
||
self.is_buy_free = True
|
||
self.is_super_free = True if buy_type == "super" else False
|
||
|
||
self.balance -= cost
|
||
self.free_spins = 0
|
||
|
||
result = {
|
||
"cost": cost,
|
||
"balance": self.balance,
|
||
"is_buy_free": self.is_buy_free,
|
||
"is_super_free": self.is_super_free,
|
||
}
|
||
return result
|
||
|
||
def doSpin(self) -> Dict:
|
||
"""
|
||
执行一次旋转/翻转操作的核心接口
|
||
|
||
:param buy_type: None=普通旋转, 'standard'=买免费旋转(100x), 'super'=买超级免费旋转(500x)
|
||
:return: 包含游戏状态和结果的字典
|
||
"""
|
||
|
||
result = {
|
||
"balance": self.balance,
|
||
"bet": self.bet,
|
||
"actual_bet": 0.0,
|
||
"is_scotter": False,
|
||
"added_spins": 0,
|
||
"win": 0.0,
|
||
"grid": self.grid,
|
||
"multipliers": self.multipliers,
|
||
"free_spins_remaining": self.free_spins,
|
||
"is_super": self.is_super_free,
|
||
"message": "",
|
||
"error": 0,
|
||
"extra_free": False,
|
||
}
|
||
|
||
# 1. 扣除下注积分
|
||
if (
|
||
not self.is_buy_free and not self.is_free_spin_status
|
||
): # 普通模式，而且不是买免费旋转
|
||
if self.balance < self.bet:
|
||
result["message"] = "游戏结束，余额不足"
|
||
result["error"] = 1
|
||
return result
|
||
result["actual_bet"] = self.bet
|
||
self.balance -= self.bet
|
||
|
||
# 初始化游戏状态
|
||
# 不是免费旋转模式则重置乘数点和累计盈利
|
||
if not self.is_free_spin_status:
|
||
self.spin_total_win = 0
|
||
self.multipliers = [[0 for _ in range(self.cols)] for _ in range(self.rows)]
|
||
|
||
# 免费旋转扣除次数
|
||
if self.free_spins > 0:
|
||
self.free_spins -= 1
|
||
|
||
# 1.2 初始化网格
|
||
if self.mock_grid:
|
||
self.grid = self.mock_grid
|
||
else:
|
||
self._init_grid()
|
||
|
||
# 2. 处理购买功能,触发>=3个的scotter
|
||
buy_free = self.is_buy_free
|
||
if self.is_buy_free:
|
||
self.is_buy_free = False
|
||
# 购买时强制触发Scatter
|
||
scatters_count = random.choices(
|
||
self.scotter_counts, weights=self.scotter_counts_weights, k=1
|
||
)[0]
|
||
self._force_scatters(int(scatters_count))
|
||
|
||
# 2.1 如果是超级免费旋转而且没有初始化，则初始化所有点为x2
|
||
if self.is_super_free and self.free_spins > 0:
|
||
self.is_super_free = False
|
||
self.multipliers = [[2 for _ in range(self.cols)] for _ in range(self.rows)]
|
||
|
||
# 3. 开始翻转循环
|
||
cascade_win = 0.0
|
||
cascade_count = 0
|
||
max_win_limit = self.bet * self.max_win_cap
|
||
|
||
steps = []
|
||
while True:
|
||
step = {
|
||
"multipler": [],
|
||
"grid": "",
|
||
# 运行后状态
|
||
"score": 0,
|
||
"symbol_links": [],
|
||
}
|
||
steps.append(step)
|
||
step["grid"] = "".join([s for row in self.grid for s in row])
|
||
for r in range(self.rows):
|
||
for c in range(self.cols):
|
||
if self.multipliers[r][c] > 0:
|
||
step["multipler"].append(r * self.cols + c)
|
||
step["multipler"].append(self.multipliers[r][c])
|
||
|
||
# 1. 查找赢奖组合
|
||
clusters = self._find_clusters()
|
||
|
||
if not clusters:
|
||
break # 没有赢奖组合，翻转结束
|
||
|
||
cascade_count += 1
|
||
current_spin_win = 0.0
|
||
|
||
# 2. 计算赔付并处理消除
|
||
# 注意：需要先收集所有要消除的位置，避免计算时网格已变
|
||
symbols_to_remove = set()
|
||
|
||
for symbol, coords in clusters:
|
||
count = len(coords)
|
||
if count < 5:
|
||
continue
|
||
|
||
# 计算该组合的乘数 (包含该区域内所有乘数点的叠加)
|
||
cluster_mult = 0
|
||
cluster_info = []
|
||
for r, c in coords:
|
||
if self.multipliers[r][c] > 1:
|
||
cluster_mult += self.multipliers[r][c]
|
||
cluster_info.append([r * 7 + c, self.multipliers[r][c]])
|
||
cluster_mult = max(1, cluster_mult)
|
||
|
||
# 基础赔付 * 符号倍率 * 组合乘数
|
||
pay_factor = get_symbol_pay(symbol, count)
|
||
win_amount = self.bet * pay_factor * cluster_mult
|
||
current_spin_win += win_amount
|
||
|
||
symbol_link = {
|
||
"symbol": symbol,
|
||
"loc": [],
|
||
"multipler": [],
|
||
"total_multi": 0,
|
||
"base_score": 0,
|
||
"score": 0,
|
||
}
|
||
symbol_link["base_score"] = self.bet * pay_factor
|
||
symbol_link["score"] = win_amount
|
||
symbol_link["total_multi"] = cluster_mult
|
||
for r, c in coords:
|
||
symbol_link["loc"].append(r * 7 + c)
|
||
for index, m in cluster_info:
|
||
symbol_link["multipler"].append(index)
|
||
symbol_link["multipler"].append(m)
|
||
step["symbol_links"].append(symbol_link)
|
||
|
||
# 记录消除位置并更新乘数点
|
||
for r, c in coords:
|
||
symbols_to_remove.add((r, c))
|
||
# 乘数点逻辑：炸开一次，点数+1 (2的幂次方)
|
||
if self.multipliers[r][c] > 0:
|
||
self.multipliers[r][c] = self.multipliers[r][c] * 2
|
||
else:
|
||
self.multipliers[r][c] = 1
|
||
|
||
if self.multipliers[r][c] > 1024:
|
||
self.multipliers[r][c] = 1024
|
||
|
||
step["score"] = current_spin_win
|
||
cascade_win += current_spin_win
|
||
|
||
# 检查是否超过最大赢利限制
|
||
if cascade_win >= max_win_limit:
|
||
cascade_win = max_win_limit
|
||
result["message"] = "达到最大赢奖上限 25000x!"
|
||
break
|
||
|
||
# 3. 执行消除和掉落
|
||
self._apply_cascade(symbols_to_remove)
|
||
|
||
# 3.1 最后处理scotter
|
||
scatter_count = sum(row.count("S") for row in self.grid)
|
||
if scatter_count >= 3:
|
||
if not self.is_free_spin_status:
|
||
self.multipliers = [[0 for _ in range(self.cols)] for _ in range(self.rows)]
|
||
self.is_free_spin_status = True
|
||
added_spins = self._add_free_spins(scatter_count)
|
||
if added_spins > 0:
|
||
result["is_scotter"] = True
|
||
result["added_spins"] = added_spins
|
||
result["message"] = f"触发免费旋转! +{added_spins}"
|
||
if added_spins > 0 and not buy_free:
|
||
result["extra_free"] = True
|
||
|
||
# 3.2 处理达到最大奖限制
|
||
raw_spin_total_win = self.spin_total_win
|
||
self.spin_total_win += cascade_win
|
||
if self.spin_total_win >= max_win_limit:
|
||
cascade_win = max_win_limit - raw_spin_total_win
|
||
self.spin_total_win = max_win_limit
|
||
self.free_spins = 0 # 立即结束
|
||
result["message"] = "达到最大赢奖上限 25000x!"
|
||
|
||
# 4. 结算
|
||
if cascade_win > 0:
|
||
self.balance += cascade_win
|
||
|
||
# 免费旋转模式结束
|
||
if self.free_spins <= 0:
|
||
self.is_free_spin_status = False
|
||
self.multipliers = [[0 for _ in range(self.cols)] for _ in range(self.rows)]
|
||
|
||
# 5. 构建结果信息
|
||
result["win"] = cascade_win
|
||
result["balance"] = self.balance
|
||
result["free_spins_remaining"] = self.free_spins
|
||
result["grid"] = self.grid
|
||
result["multipliers"] = self.multipliers
|
||
result["steps"] = steps
|
||
result["spin_total_win"] = self.spin_total_win
|
||
return result
|
||
|
||
def _find_clusters(self) -> Dict[str, List[Tuple[int, int]]]:
|
||
"""查找所有相连的符号块 (排除SCATTER)"""
|
||
visited = set()
|
||
clusters = []
|
||
|
||
for r in range(self.rows):
|
||
for c in range(self.cols):
|
||
symbol = self.grid[r][c]
|
||
if symbol == "S" or (r, c) in visited:
|
||
continue
|
||
|
||
# BFS 查找连通块
|
||
queue = [(r, c)]
|
||
connected = []
|
||
while queue:
|
||
curr_r, curr_c = queue.pop(0)
|
||
if (curr_r, curr_c) in visited:
|
||
continue
|
||
if self.grid[curr_r][curr_c] == symbol:
|
||
visited.add((curr_r, curr_c))
|
||
connected.append((curr_r, curr_c))
|
||
# 检查上下左右
|
||
for dr, dc in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
|
||
nr, nc = curr_r + dr, curr_c + dc
|
||
if 0 <= nr < self.rows and 0 <= nc < self.cols:
|
||
if (nr, nc) not in visited and self.grid[nr][
|
||
nc
|
||
] == symbol:
|
||
queue.append((nr, nc))
|
||
|
||
if len(connected) >= 5:
|
||
clusters.append((symbol, connected))
|
||
|
||
return clusters
|
||
|
||
def _apply_cascade(self, remove_coords: set):
|
||
"""执行消除和掉落逻辑"""
|
||
# 1. 消除
|
||
for r, c in remove_coords:
|
||
self.grid[r][c] = None # 标记为空
|
||
|
||
# 2. 掉落 (按列处理)
|
||
for c in range(self.cols):
|
||
# 提取该列所有非空符号
|
||
existing_syms = [
|
||
self.grid[r][c] for r in range(self.rows) if self.grid[r][c] is not None
|
||
]
|
||
# 填充新符号到顶部
|
||
needed = self.rows - len(existing_syms)
|
||
new_syms = self._get_next_drop_symbol(
|
||
count=needed, no_scatter=self.grid_col_has_scotter(c)
|
||
)
|
||
# 重新组合该列 (新符号在上，原有符号在下)
|
||
new_col = new_syms + existing_syms
|
||
|
||
# 更新回网格
|
||
for r in range(self.rows):
|
||
self.grid[r][c] = new_col[r]
|
||
|
||
def _force_scatters(self, count: int):
|
||
"""强制在随机位置放置指定数量的Scatter"""
|
||
# 检查是否已经有，有的话改成其他符号
|
||
for r in range(self.rows):
|
||
for c in range(self.cols):
|
||
if self.grid[r][c] == "S":
|
||
self.grid[r][c] = "G"
|
||
cols = list(range(self.cols))
|
||
for _ in range(count):
|
||
x = random.choice(cols)
|
||
y = random.randint(0, self.rows - 1)
|
||
self.grid[y][x] = "S"
|
||
cols.remove(x)
|
||
|
||
def _add_free_spins(self, scatter_count: int) -> int:
|
||
"""根据Scatter数量增加免费旋转次数"""
|
||
if scatter_count < 3:
|
||
return 0
|
||
spins = scotter_count_mapping.get(str(scatter_count), 30)
|
||
self.free_spins += spins
|
||
return spins
|
||
|
||
|
||
def no_cluster_grid():
|
||
return [
|
||
(
|
||
["B" if col % 2 == 0 else "E" for col in range(7)]
|
||
if row % 2 == 0
|
||
else ["E" if col % 2 == 0 else "B" for col in range(7)]
|
||
)
|
||
for row in range(7)
|
||
]
|
||
|
||
|
||
if __name__ == "__main__":
|
||
begin_balance = 1_0000_0000
|
||
custom_grid = no_cluster_grid()
|
||
custom_grid[0] = ["A"] * 7 # 设置一个乘数点
|
||
game = SugarRush1000(
|
||
balance=begin_balance,
|
||
weights={
|
||
"A": 11.910798577481279,
|
||
"B": 23.10218485800891,
|
||
"C": 36.43351025439602,
|
||
"D": 46.62768084621008,
|
||
"E": 54.778674468859805,
|
||
"F": 64.18383150145051,
|
||
"G": 75.00641730395597,
|
||
"S": 1.5190499131843918,
|
||
},
|
||
scotter_counts_weights={
|
||
"3": 7.26421894353717,
|
||
"4": 4.1724734692682395,
|
||
"5": 0.8119106579617028,
|
||
"6": 0.20313929837878217,
|
||
"7": 0.04857599989214818,
|
||
},
|
||
mock_grid=custom_grid,
|
||
)
|
||
|
||
game.drop_sequence = [
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"C",
|
||
"A",
|
||
"B",
|
||
"C",
|
||
"D",
|
||
"A",
|
||
"B",
|
||
] # 预设掉落全是 C
|
||
|
||
res = game.doSpin()
|
||
print(json.dumps(res))
|