写一个智能的程序来玩德州扑克，是否可行？

1import time 2from itertools import combinations 3import matplotlib.pyplot as plt 4import numpy as np 5import poker 6from phevaluator import _evaluate_cards, sample_cards, card 7import pandas as pd 8import seaborn as sns 9import sqlite3 10import torch 11import sys 12sys.path.append('../decisionmaker') 13import evaluator 14from scipy import interpolate 15import math

1# 加载策略 2conn = sqlite3.connect('../example.db') 3cur = conn.cursor() 4str_3b_query = cur.execute("select strategy_numpy from strategy where strategy_name = 'BTN_2.5strd_STR_strategy'").fetchone() 5btn_call_3b_query = cur.execute("select strategy_numpy from strategy where strategy_name = 'BTN_2.5strd_STR_11.0strd_BTN_strategy'").fetchone() 6str_3b_numpy = np.frombuffer(str_3b_query[0]).reshape((13, 13, 3)) 7btn_call_3b_numpy = np.frombuffer(btn_call_3b_query[0]).reshape((13, 13, 3))

1# str位置面对btn open AA的 raise allin call 频率 2print(str_3b_numpy.shape) 3print(str_3b_numpy[0,:5]) 1(13, 13, 3) 2[[1. 0. 0. ] 3 [1. 0. 0. ] 4 [1. 0. 0. ] 5 [0.588 0. 0.412] 6 [0.148 0. 0.852]]

1# 手牌的index可视化 2def card_index_to_human(*index): 3 r = ['2', '3', '4', '5', '6', '7', '8', '9', 'T', 'J', 'Q', 'K', 'A'] 4 s = ['c', 'd', 'h', 's'] 5 result = [r[i // 4] + s[i % 4] for i in sorted(index, reverse=True)] 6 return "".join(result) 7 8# 手牌范围的numpy格式转换为每一个组合 如AKs就转换为 AcKc AdKd AsKs AhKh四种组合 返回一个pd.df 9# 第一列为手牌组合 第一列为扑克点数 之后为每个行动的频率 10def range_numpy_to_combo_list(player_range: np.ndarray): 11 # shape 13 13 3 12 hand_list = [] 13 for row in range(13): 14 for col in range(13): 15 first_index = 4 * (12 - row) 16 first_list = list(range(first_index, first_index + 4)) 17 second_index = 4 * (12 - col) 18 second_list = list(range(second_index, second_index + 4)) 19 if row == col: 20 # pair 21 comb_list = list(combinations(range(4 * (12 - row), 4 * (12 - row) + 4), 2)) 22 comb_list = [(tuple(sorted(t)), card_index_to_human(*t), player_range[row][col][0], player_range[row][col][1], player_range[row][col][2]) for t in comb_list] 23 hand_list = hand_list + comb_list 24 elif row > col: 25 # offsuit 26 tmp_list = [] 27 for i in range(4): 28 for j in range(4): 29 if i != j: 30 tmp_list.append((tuple(sorted((first_list[i], second_list[j]))), card_index_to_human(*(first_list[i], second_list[j])), player_range[row][col][0], player_range[row][col][1], player_range[row][col][2])) 31 hand_list = hand_list + tmp_list 32 else: 33 # suit 34 tmp_list = [] 35 for i in range(4): 36 for j in range(4): 37 if i == j: 38 tmp_list.append((tuple(sorted((first_list[i], second_list[j]))), card_index_to_human(*(first_list[i], second_list[j])), player_range[row][col][0], player_range[row][col][1], player_range[row][col][2])) 39 hand_list = hand_list + tmp_list 40 res = pd.DataFrame(hand_list, columns = ['hand', 'human', 'raise', 'allin', 'call']) 41 sorted_df = res.sort_values(by='hand', key=lambda x: x.apply(lambda y: (y[0], y[1]))).reset_index(drop=True) 42 return sorted_df

1%%time 2str_3b_combo_df = range_numpy_to_combo_list(str_3b_numpy) 3btn_call_3b_combo_df = range_numpy_to_combo_list(btn_call_3b_numpy) 4print(str_3b_combo_df.shape) 5str_3b_combo_df[str_3b_combo_df['raise'] > 0][:5] 1(1326, 5) 2CPU times: total: 0 ns 3Wall time: 14.6 ms

1%%time 2 3# 根据 flop 的牌面到库里拿到 flop equity matrix 4def get_equity_matrix_from_db(board: tuple): 5 conn = sqlite3.connect(r"E:\project_p\preflop_tips\flop_equity.db") 6 cur = conn.cursor() 7 board = "-".join([str(x) for x in board]) 8 query = cur.execute(f"select flop, equity_matrix, mask_matrix from flop_equity where flop = '{board}'").fetchone() 9 conn.close() 10 equity_matrix = np.frombuffer(query[1], dtype=np.float32).reshape((1326, 1326)) 11 mask_matrix = np.frombuffer(query[2], dtype=bool).reshape((1326, 1326)) 12 13 return equity_matrix, mask_matrix 14 15ALL_COMBINATION = list(combinations(range(52), 2)) 16def get_hand_index(hand: tuple): 17 hand = tuple(sorted(hand)) 18 return ALL_COMBINATION.index(hand) 1CPU times: total: 0 ns 2Wall time: 0 ns

1%%time 2# 计算单手牌的胜率 3def flop_hand_vs_range_equity(board: tuple, player_1_hand: tuple, player_2_strategy_df: pd.DataFrame, player_2_strategy: str): 4 equity_matrix, mask_matrix = get_equity_matrix_from_db(board) 5 player_1_hand_index = get_hand_index(player_1_hand) 6 player_2_strategy_df['proportion'] = player_2_strategy_df[player_2_strategy] / player_2_strategy_df[player_2_strategy][mask_matrix[player_1_hand_index]].sum() 7 player_2_strategy_df['proportion'] = np.nan_to_num(player_2_strategy_df['proportion'], nan=0.0) 8 9 return np.sum(equity_matrix[player_1_hand_index] * player_2_strategy_df['proportion'].values) 10flop_hand_vs_range_equity((29,34,42), (50, 51), btn_call_3b_combo_df, 'call') 1CPU times: total: 15.6 ms 2Wall time: 54.2 ms 3 4 5 6 7 80.7042426706000064

1# 范围对上范围的胜率 player_1_df新增胜率列 2def flop_range_vs_range_equity(board, player_1_strategy_df: pd.DataFrame, player_1_strategy: str, player_2_strategy_df: pd.DataFrame, player_2_strategy: str, *matrix): 3 player_1_strategy_df = player_1_strategy_df.sort_index() 4 player_2_strategy_df = player_2_strategy_df.sort_index() 5 6 if matrix: 7 equity_matrix, mask_matrix = matrix 8 else: 9 equity_matrix, mask_matrix = get_equity_matrix_from_db(board) 10 # 每手牌的权重打横之后拓展到1326行 11 p1_repeat = np.repeat(player_1_strategy_df[player_1_strategy].values.reshape(1,-1),1326,axis=0) 12 p2_repeat = np.repeat(player_2_strategy_df[player_2_strategy].values.reshape(1,-1),1326,axis=0) 13 # 权重乘以mask矩阵剔除干涉手牌 14 p1_freq = mask_matrix * p1_repeat 15 p2_freq = mask_matrix * p2_repeat 16 # 每一行权重求和 17 p1_freq_sum = np.sum(p1_freq, axis=1) 18 p2_freq_sum = np.sum(p2_freq, axis=1) 19 # 找出为0的index 避免除0 20 p1_no_zero = np.where(p1_freq_sum != 0) 21 p2_no_zero = np.where(p2_freq_sum != 0) 22 # 构建全0结果矩阵 23 p1_result = np.zeros((1326,)) 24 p2_result = np.zeros((1326,)) 25 # 求p1对p2的整体胜率 用equity_matrix乘以p2_freq再按行求和 26 equity_matrix_1_vs_2 = np.sum(equity_matrix * p2_freq, axis=1) 27 equity_matrix_2_vs_1 = np.sum(equity_matrix * p1_freq, axis=1) 28 29 p1_result[p2_no_zero] = equity_matrix_1_vs_2[p2_no_zero] / p2_freq_sum[p2_no_zero] 30 p2_result[p1_no_zero] = equity_matrix_2_vs_1[p1_no_zero] / p1_freq_sum[p1_no_zero] 31 32 player_1_strategy_df['equity'] = p1_result 33 player_2_strategy_df['equity'] = p2_result 34 35 player_1_strategy_df['proportion'] = 100 * (np.sum(p1_freq, axis=0) / np.sum(np.sum(p1_freq, axis=0))) 36 player_2_strategy_df['proportion'] = 100 * (np.sum(p2_freq, axis=0) / np.sum(np.sum(p2_freq, axis=0))) 37 38 player_1_strategy_df = player_1_strategy_df[player_1_strategy_df['proportion'] > 0].sort_values(by='equity', ascending=False) 39 player_2_strategy_df = player_2_strategy_df[player_2_strategy_df['proportion'] > 0].sort_values(by='equity', ascending=False) 40 41 player_1_strategy_df['cumsum'] = player_1_strategy_df['proportion'].cumsum() 42 player_2_strategy_df['cumsum'] = player_2_strategy_df['proportion'].cumsum() 43 44 # 构造cumsum 为0 y与equity max一致 45 player_1_strategy_df = pd.concat([pd.DataFrame({'cumsum': [0], 'equity': [player_1_strategy_df.iloc[0]['equity']]}), player_1_strategy_df], ignore_index=True) 46 player_2_strategy_df = pd.concat([pd.DataFrame({'cumsum': [0], 'equity': [player_2_strategy_df.iloc[0]['equity']]}), player_2_strategy_df], ignore_index=True) 47 48 49 return player_1_strategy_df, player_2_strategy_df

1%%time 2str_3b_equity_df, btn_call_3b_equity_df = flop_range_vs_range_equity((29,34,42), str_3b_combo_df, 'raise', btn_call_3b_combo_df, 'call') 3str_3b_equity_df.head() 1CPU times: total: 62.5 ms 2Wall time: 69.2 ms

1%%time 2CARD_COMBO = list(combinations(range(52), 2)) 3 4# 绘制equity曲线并标记当前手牌的位置 5def paint_equity_curve(p1_df, p2_df, *p1_hand): 6 plt.figure(figsize=(8, 6)) # 设置画布的大小 7 sns.set_palette("hls") # 设置所有图的颜色，使用hls色彩空间 8 9 plt.xlabel('Hand_rank', fontsize=15) # 添加x轴标签，并改变字体 10 plt.ylabel('Equity', fontsize=15) # 添加y轴变浅，并改变字体 11 12 plt.grid(line style='-') # 添加网格线 13 14 sns.lineplot(p1_df, x='cumsum', y='equity', color=sns.color_palette("RdBu")[4]) 15 sns.lineplot(p2_df, x='cumsum', y='equity', color=sns.color_palette("RdBu")[0]) 16 sns.despine(ax=None, top=True, right=True, left=True, bottom=True) # 将图像的框框删掉 17 if p1_hand: 18 hand_row = p1_df[p1_df['hand'] == tuple(sorted(p1_hand[0]))] 19 plt.scatter(hand_row['cumsum'], hand_row['equity'], color='red', marker='x') 20 21 plt.show() 22paint_equity_curve(str_3b_equity_df, btn_call_3b_equity_df, (50, 51))

1%%time 2# equity差值函数 3def cal_equity_diff(p1_df, p2_df): 4 p1_df_interpolate_func = interpolate.interp1d(p1_df['cumsum'], p1_df['equity']) 5 p2_df_interpolate_func = interpolate.interp1d(p2_df['cumsum'], p2_df['equity']) 6 return p1_df_interpolate_func(list(range(100))) - p2_df_interpolate_func(list(range(100))) 7 8 9# 绘制equity_diff曲线 10def paint_equity_diff_curve(p1_df, p2_df, p1_hand=None): 11 12 diff = cal_equity_diff(p1_df, p2_df) 13 plt.figure(figsize=(8, 6)) # 设置画布的大小 14 if p1_hand: 15 is_have_hand = p1_df['hand'].loc[1:] == tuple(sorted(p1_hand)) 16 if any(is_have_hand): 17 hand_row = str_3b_equity_df.loc[1:][str_3b_equity_df['hand'].loc[1:] == tuple(sorted((50,51)))] 18 plt.scatter(hand_row['cumsum'], diff[int(hand_row['cumsum'].iloc[0])], color=sns.color_palette("RdBu")[0], marker='x') 19 20 21 sns.set_palette("hls") 22 plt.xlabel('Hand_rank', fontsize=15) 23 plt.ylabel('Equity_diff', fontsize=15) 24 plt.axhline(y=0, color=sns.color_palette("RdBu")[0], line style='--') 25 plt.grid(line style='-') # 添加网格线 26 sns.lineplot(diff, color=sns.color_palette("RdBu")[4]) 27 sns.despine(ax=None, top=True, right=True, left=True, bottom=True) # 将图像的框框删掉 28 29 30 plt.show() 31hand = (50, 51) 32paint_equity_diff_curve(str_3b_equity_df, btn_call_3b_equity_df, hand)

1def evaluate_cards_input_torch(cards_tc): 2 hand_size = cards_tc.shape[1] 3 no_flush = NO_FLUSHES[hand_size] 4 hand_count = cards_tc.shape[0] 5 6 # suit_hash = torch.sum(SUITBIT_BY_ID[cards_tc], axis=1) 7 suit_hash = torch.sum(1 << (cards_tc % 4 * 3), axis=1) 8 flush_suit = (SUITS[suit_hash] - 1).reshape((hand_count, 1)) 9 10 flush_suit_mask_np = torch.where(flush_suit != -1, True, False).reshape((hand_count,)) 11 # 找到 card % 4 == flush_suit 这个条件的index 打标为True 12 ############################################# 13 # FLUSH 14 cards_suit = cards_tc % 4 15 # [34, 30, 14, 22, 32]] 16 mask = torch.where(cards_suit - flush_suit == 0, 0x1 << (cards_tc // 4), 0) # n x 5 17 mask_t = mask.t() 18 hand_binary = mask_t[0].to(dtype=torch.int) 19 for i in range(1, hand_size): 20 hand_binary = torch.bitwise_or(hand_binary, mask_t[i]) 21 hand_binary = hand_binary.t().reshape((hand_count, 1)) 22 flush_result = FLUSH[hand_binary].reshape((hand_count,)) 23 ############################################# 24 # NO FLUSH 25 # ------------------------------------------- 26 cards_rank = cards_tc >> 2 27 cards_rank_t = cards_rank.t() 28 row_counts = torch.zeros((13, cards_rank.shape[0]), dtype=torch.int8, device='cuda:0') 29 rg = torch.arange(cards_rank.shape[0], dtype=torch.int, device='cuda:0') 30 for rank in cards_rank_t: 31 row_counts[rank.to(dtype=torch.int), rg] = row_counts[rank.to(dtype=torch.int), rg] + 1 32 # ------------------------------------------- 33 # 转置之后row为当前rank点数所有手牌组合的array 0 [1 0 1 1 1 1] 34 sum_numb = torch.zeros((1, row_counts.shape[1]), dtype=torch.int, device='cuda:0') 35 num_cards = torch.full((row_counts.shape[1],), hand_size, device='cuda:0') 36 full_tc = torch.full((row_counts.shape[1],), 12, dtype=torch.int, device='cuda:0') 37 for rank, row in enumerate(row_counts): 38 sum_numb = sum_numb + DP[row.to(dtype=torch.int), full_tc - rank, num_cards] 39 num_cards = num_cards - row 40 result = no_flush[sum_numb.t().reshape((hand_count,))] 41 flush = flush_result * flush_suit_mask_np 42 non_flush = result * (~flush_suit_mask_np) 43 return flush + non_flush 44 45def evaluate_fix_flop_turn_matrix(flop: tuple, player_range_1: [tuple], player_range_2: [tuple]): 46 """ 47 转牌一共45张出牌,再遍历河牌44张出牌计算固定转牌下的equity matrix 48 :param flop:index tuple (12, 24, 37) 49 :param player_range_1: sorted list index tuple [(19, 26), (19, 27)...] 50 :param player_range_2: sorted list index tuple [(35, 47), (35, 48)...] 51 :return: turn card list [1, 2, 3], ever turn card equity matrix 52 """ 53 # 两个玩家手牌组合的笛卡尔积 54 player_range_cartesian = [lh + rh for lh in player_range_1 for rh in player_range_2] 55 player_range_cartesian_size = len(player_range_cartesian) 56 # 方便找到(13, 45)这种手牌组合在1326个组合中的index 57 helper_index = torch.tensor(reduce(lambda x, y: x + [x[-1] + y], list(range(51, 0, -1)), [0]), dtype=torch.int, 58 device='cuda:0') 59 60 player_hands_tc = torch.tensor(player_range_cartesian, dtype=torch.int8, device='cuda:0').repeat(1, 49) \ 61 .reshape(49 * player_range_cartesian_size, -1) 62 63 # 剔除掉flop3张后剩余的牌堆 64 exclude_flop_card = sorted(tuple(set(range(52)) - set(flop))) 65 66 # 构建flop river的固定组合 都是 player_range_cartesian_size * 49 行 67 river_temp_tc = torch.tensor(exclude_flop_card, dtype=torch.int8, device='cuda:0').reshape((49, 1)) 68 river_tc = river_temp_tc.repeat(player_range_cartesian_size, 1) 69 flop_tc = torch.tensor(flop, dtype=torch.int8, device='cuda:0').repeat(49 * player_range_cartesian_size, 1) 70 71 # 返回 turn 和 matrix list 72 result_turn_list = [] 73 result_equity_matrix_list = [] 74 result_mask_matrix_list = [] 75 for turn_card in exclude_flop_card: 76 result_turn_list.append(turn_card) 77 turn_tc = torch.tensor(turn_card, dtype=torch.int8, device='cuda:0').repeat(49 * player_range_cartesian_size, 1) 78 # 先不考虑手牌干涉 合并之后再剔除干涉组合 79 all_board = torch.cat((flop_tc, player_hands_tc, turn_tc, river_tc), dim=1) 80 # 剔除干涉的行 81 row_counts = torch.zeros((52, all_board.shape[0]), dtype=torch.int8, device='cuda:0') 82 rg = torch.arange(all_board.shape[0], dtype=torch.int, device='cuda:0') 83 for rank in all_board.t(): 84 row_counts[rank.to(dtype=torch.int), rg] = row_counts[rank.to(dtype=torch.int), rg] + 1 85 # 剔除干涉后hand vs hand river一共有44种出牌 86 interference_mask = ~(torch.max(row_counts, dim=0).values > 1) # 无效置false 87 all_board = all_board[interference_mask] 88 89 player_rank_1 = evaluate_cards_input_torch(all_board[:, [0, 1, 2, 3, 4, 7, 8]]) 90 player_rank_2 = evaluate_cards_input_torch(all_board[:, [0, 1, 2, 5, 6, 7, 8]]) 91 92 # 每一行为一个组合对上一个组合的胜率 93 single_turn_greater_equity = torch.sum(((player_rank_1 - player_rank_2) < 0).reshape(-1, 44), dim=1) / 44 94 single_turn_equal_equity = 0.5 * torch.sum(((player_rank_1 - player_rank_2) == 0).reshape(-1, 44), dim=1) / 44 95 96 single_turn_greater_equity_t = torch.sum(((player_rank_2 - player_rank_1) < 0).reshape(-1, 44), dim=1) / 44 97 single_turn_equal_equity_t = 0.5 * torch.sum(((player_rank_2 - player_rank_1) == 0).reshape(-1, 44), dim=1) / 44 98 del player_rank_1 99 del player_rank_2 100 base_matrix = torch.zeros((1326, 1326), dtype=torch.float32, device='cuda:0') 101 mask_matrix = torch.zeros((1326, 1326), dtype=torch.int8, device='cuda:0') 102 103 all_board = all_board.reshape(-1, 44, 9) 104 105 base_matrix[ 106 helper_index[all_board[:, 0, 3].to(dtype=torch.int)] + all_board[:, 0, 4] - all_board[:, 0, 3] - 1, 107 helper_index[all_board[:, 0, 5].to(dtype=torch.int)] + all_board[:, 0, 6] - all_board[:, 0, 5] - 1] \ 108 = single_turn_greater_equity + single_turn_equal_equity 109 base_matrix[ 110 helper_index[all_board[:, 0, 5].to(dtype=torch.int)] + all_board[:, 0, 6] - all_board[:, 0, 5] - 1, 111 helper_index[all_board[:, 0, 3].to(dtype=torch.int)] + all_board[:, 0, 4] - all_board[:, 0, 3] - 1] \ 112 = single_turn_greater_equity_t + single_turn_equal_equity_t 113 mask_matrix[ 114 helper_index[all_board[:, 0, 3].to(dtype=torch.int)] + all_board[:, 0, 4] - all_board[:, 0, 3] - 1, 115 helper_index[all_board[:, 0, 5].to(dtype=torch.int)] + all_board[:, 0, 6] - all_board[:, 0, 5] - 1] \ 116 = 1 117 mask_matrix[ 118 helper_index[all_board[:, 0, 5].to(dtype=torch.int)] + all_board[:, 0, 6] - all_board[:, 0, 5] - 1, 119 helper_index[all_board[:, 0, 3].to(dtype=torch.int)] + all_board[:, 0, 4] - all_board[:, 0, 3] - 1] \ 120 = 1 121 122 result_equity_matrix_list.append(base_matrix) 123 result_mask_matrix_list.append(mask_matrix) 124 125 return result_turn_list, result_equity_matrix_list, result_mask_matrix_list

1%%time 2turn_list, equity_matrix_list, mask_matrix_list = evaluator.evaluate_fix_flop_turn_matrix((29, 34, 42), str_3b_equity_df.sort_index()['hand'][1:], btn_call_3b_equity_df.sort_index()['hand'][1:]) 1CPU times: total: 2.73 s 2Wall time: 2.91 s 1%%time 2p1_stg_df_list = [] 3p2_stg_df_list = [] 4 5for index, turn_card in enumerate(turn_list): 6 p1_stg_df, p2_stg_df = flop_range_vs_range_equity((29, 34, 42), str_3b_combo_df, 'raise', btn_call_3b_combo_df, 'call',*(equity_matrix_list[index].cpu().numpy(), mask_matrix_list[index].cpu().numpy())) 7 p1_stg_df['turn'] = turn_card 8 p2_stg_df['turn'] = turn_card 9 p1_stg_df_list.append(p1_stg_df) 10 p2_stg_df_list.append(p2_stg_df) 1CPU times: total: 1.36 s 2Wall time: 1.61 s

1hand = (50, 51) 2fig,axes=plt.subplots(13,4, figsize=(24, 49)) 3 4for index in range(49): 5 if hand[0] != turn_list[index] and hand[1] != turn_list[index]: 6 diff = cal_equity_diff(p1_stg_df_list[index], p2_stg_df_list[index]) 7 sns.set_palette("hls") # 设置所有图的颜色，使用hls色彩空间 8 hand_row = p1_stg_df_list[index].loc[1:][p1_stg_df_list[index]['hand'].loc[1:] == tuple(sorted(hand))] 9 sns.scatterplot( x=hand_row['cumsum'], y=(diff[math.ceil(hand_row['cumsum'].iloc[0])] + diff[math.ceil(hand_row['cumsum'].iloc[0]) - 1]) / 2,color=sns.color_palette("RdBu")[0],ax=axes[turn_list[index]//4, turn_list[index]%4]) 10 sns.lineplot(diff, color=sns.color_palette("RdBu")[4], ax=axes[turn_list[index]//4, turn_list[index]%4]).set_title(card_index_to_human(turn_list[index])) 11 axes[turn_list[index]//4, turn_list[index]%4].axhline(y=0, color=sns.color_palette("RdBu")[0], line style='--') 12 axes[turn_list[index]//4, turn_list[index]%4].set_xlabel("") 13 axes[turn_list[index]//4, turn_list[index]%4].vlines(x=hand_row['cumsum'], ymin=0, ymax=max(diff), color="red") 14 sns.despine(ax=None, top=True, right=True, left=True, bottom=True) # 将图像的框框删掉 15 16 17plt.tight_layout() 18plt.show()