import itertools
from collections import deque
# ===== 可配置参数 =====
LONG_TERM_WINDOW = 200 # 长期分析期数
SHORT_TERM_WINDOW = 10 # 短期分析期数
RECOMMEND_DIGITS_PER_POSITION = 6 # 每位推荐数字数（6→6×6×6=216注）
HOT_DIGIT_THRESHOLD = 0.3 # 出现频率>30%视为高频数字
MATRIX_DECAY_FACTOR = 0.7 # 转移矩阵衰减因子
# =====================
# 初始化历史数据
history_long = deque(maxlen=LONG_TERM_WINDOW)
history_short = deque(maxlen=SHORT_TERM_WINDOW)
# 示例数据
initial_data = [
'444','523','195','216','404',
'347','701','143','841','831','324','801','395','436','812',
'274','681','088','623','134','886','859','465','905','285',
'034','347','156','945','810','006','004','340','030','044',
'701','552','898','037','680','581','933','877','398','201',
'425','841','508','582','843','483','718','533','413','471',
'080','071','332','355','243','793','863','050','678','914',
'630','023','257','134','125','147','811','712','679','848',
'746','499','249','172','155','918','031','653','084','481',
'546','620','677','893','192','907','220','788','691','123',
'900','168','351','617','656','362','128','806','143','027',
'471','978','508','398','963','956','292','734','159','346',
'484','788','228','677','346','048','822','489','297','513',
'392','934','309','849','521','578','366','588','875','916',
'675','552','805','702','372','806','570','037','570','973',
'337','899','384','916','716','765','672','771','793','490',
'257','466','656','990','184','671','975','934','771','462',
'935','132','179','743','407','425','113','296','738','766',
'344','593','304','353','634','845','222','208','993','830',
'721','432','534','672','948','171','192','174','997','911',
'233','471','144','084','076','726',
]
history_long.extend(initial_data)
history_short.extend(initial_data[-SHORT_TERM_WINDOW:])
# 初始化转移矩阵
transfer_bai_long = [[0]*10 for _ in range(10)]
transfer_shi_long = [[0]*10 for _ in range(10)]
transfer_ge_long = [[0]*10 for _ in range(10)]
transfer_bai_short = [[0]*10 for _ in range(10)]
transfer_shi_short = [[0]*10 for _ in range(10)]
transfer_ge_short = [[0]*10 for _ in range(10)]
def detect_hot_digits(counts_list):
"""动态识别高频数字"""
total = max(1, sum(counts_list))
return {i for i, cnt in enumerate(counts_list) if cnt/total > HOT_DIGIT_THRESHOLD}
def _update_single_matrix(matrix, from_digit, to_digit, hot_digits):
"""增强版矩阵更新"""
from_d = int(from_digit)
to_d = int(to_digit)
boost = 1.5 if to_d in hot_digits else 1.0 # 高频数字权重提升50%
matrix[from_d][to_d] = matrix[from_d][to_d] * MATRIX_DECAY_FACTOR + boost
def update_transfer_matrices(new_number):
"""更新转移矩阵"""
short_bai, short_shi, short_ge = calculate_counts(history_short)
hot_digits = detect_hot_digits(short_bai) | detect_hot_digits(short_shi) | detect_hot_digits(short_ge)
if len(history_long) >= 2:
last_num = history_long[-2]
curr_num = history_long[-1]
_update_single_matrix(transfer_bai_long, last_num[0], curr_num[0], hot_digits)
_update_single_matrix(transfer_shi_long, last_num[1], curr_num[1], hot_digits)
_update_single_matrix(transfer_ge_long, last_num[2], curr_num[2], hot_digits)
if len(history_short) >= 2:
last_num = history_short[-2]
curr_num = history_short[-1]
_update_single_matrix(transfer_bai_short, last_num[0], curr_num[0], hot_digits)
_update_single_matrix(transfer_shi_short, last_num[1], curr_num[1], hot_digits)
_update_single_matrix(transfer_ge_short, last_num[2], curr_num[2], hot_digits)
def get_top_numbers(matrix, last_digit, global_counts, short_counts):
"""统一获取推荐数字（核心修正点）"""
# 长期推荐
long_row = matrix[last_digit]
long_total = max(1, sum(long_row))
long_probs = sorted([(i, c/long_total) for i,c in enumerate(long_row) if c>0],
key=lambda x: (-x[1], -global_counts[x[0]]))
# 短期推荐
short_row = [sum(col) for col in zip(*[m[last_digit] for m in
[transfer_bai_short, transfer_shi_short, transfer_ge_short]])]
short_total = max(1, sum(short_row))
short_probs = sorted([(i, c/short_total) for i,c in enumerate(short_row) if c>0],
key=lambda x: (-x[1], -short_counts[x[0]]))
# 合并结果
combined = []
for lst in [long_probs, short_probs]:
combined.extend([d for d,_ in lst[:RECOMMEND_DIGITS_PER_POSITION//2]])
# 去重并补足
result = list(dict.fromkeys(combined))
if len(result) < RECOMMEND_DIGITS_PER_POSITION:
global_top = sorted(range(10), key=lambda x: -global_counts[x])
result.extend(d for d in global_top if d not in result)
return result[:RECOMMEND_DIGITS_PER_POSITION]
def calculate_counts(data):
"""计算数字频率"""
counts = ([0]*10, [0]*10, [0]*10)
for num in data:
counts[0][int(num[0])] += 1
counts[1][int(num[1])] += 1
counts[2][int(num[2])] += 1
return counts
def generate_recommendations():
"""生成推荐号码"""
if len(history_long) < 2:
print("需要至少2期历史数据")
return
last_number = history_long[-1]
print(f"上期号码: {last_number}")
# 计算频率
global_bai, global_shi, global_ge = calculate_counts(history_long)
short_bai, short_shi, short_ge = calculate_counts(history_short)
# 获取推荐
bai_rec = get_top_numbers(transfer_bai_long, int(last_number[0]), global_bai, short_bai)
shi_rec = get_top_numbers(transfer_shi_long, int(last_number[1]), global_shi, short_shi)
ge_rec = get_top_numbers(transfer_ge_long, int(last_number[2]), global_ge, short_ge)
# 生成组合
numbers = list(itertools.product(bai_rec, shi_rec, ge_rec))
print(f"\n推荐{RECOMMEND_DIGITS_PER_POSITION}×{RECOMMEND_DIGITS_PER_POSITION}×{RECOMMEND_DIGITS_PER_POSITION}={len(numbers)}注:")
for i, num in enumerate(numbers, 1):
print(f"{num[0]}{num[1]}{num[2]}", end=" \n" if i%20==0 else " ")
def add_new_number(new_number):
"""添加新号码"""
if len(new_number) != 3 or not new_number.isdigit():
print("需3位数字")
return
history_long.append(new_number)
history_short.append(new_number)
update_transfer_matrices(new_number)
generate_recommendations()
if __name__ == "__main__":
print("初始分析:")
generate_recommendations()
# 测试案例
test_nums = []
for n in test_nums:
print(f"\n{'='*30}\n新增号码: {n}")
add_new_number(n)
