import pandas as pd import numpy as np import json import configparser from datetime import datetime, timedelta conf = configparser.ConfigParser() # 读取配置文件 conf.read('conf.ini', encoding='utf-8') class LotteryPredictorV14Profit: """ V28 "Iron-Shield" (铁盾终极版) 核心逻辑: V27 动态流水增强 (3-8注) + 绝佳 30% (6000元) 日内止损熔断。 目标: 100元/注重仓下，最高收益/回撤比。 """ # --- 核心参数 --- ODDS = float(conf["init"]["ODDS"]) REBATE_RATE = float(conf["dynamic"]["REBATE_RATE"]) BASE_BET_UNIT = int(conf["init"]["BASE_BET_UNIT"]) _bet_this_round = int(conf["init"]["BET_THIS_ROUND"]) SAFE_BET_LEVEL_1 = 0.05 SAFE_BET_LEVEL_2 = 0.10 SAFE_BET_LEVEL_3 = 0.15 SAFE_BET_LEVEL_4 = 0.20 SAFE_BET_LEVEL_5 = 0.25 RADICAL_BET_LEVEL_0 = 0.05 RADICAL_BET_LEVEL_1 = 0.1 RADICAL_BET_LEVEL_2 = 0.2 INITIAL_CAPITAL = int(conf["init"]["INITIAL_CAPITAL"]) DAILY_STOP_LOSS_RATE = float(conf["init"]["DAILY_STOP_LOSS_RATE"]) # 绝佳熔断点：6000元熔断 POSITION_LOOKBACK_PERIODS = int(conf["dynamic"]["POSITION_LOOKBACK_PERIODS"]) OMISSIONS_LEVEL_1 = int(conf["dynamic"]["OMISSIONS_LEVEL_1"]) OMISSIONS_LEVEL_2 = int(conf["dynamic"]["OMISSIONS_LEVEL_2"]) OMISSIONS_LEVEL_3 = int(conf["dynamic"]["OMISSIONS_LEVEL_3"]) MOMENTUM_LEVEL_1 = int(conf["dynamic"]["MOMENTUM_LEVEL_1"]) MOMENTUM_LEVEL_2 = int(conf["dynamic"]["MOMENTUM_LEVEL_2"]) MOMENTUM_LEVEL_3 = int(conf["dynamic"]["MOMENTUM_LEVEL_3"]) RECENT_OMISSIONS = int(conf["dynamic"]["RECENT_OMISSIONS"]) RECENT_MOMENTUM = int(conf["dynamic"]["RECENT_MOMENTUM"]) # V27 分级投注参数 # S级: 遗漏>=55, 动能>=45 -> 8注 # A级: 遗漏>=50, 动能>=40 -> 6注 # B级: 遗漏>=45, 动能>=35 -> 4注 def __init__(self): # 用于追踪当日盈亏，确保日内止损 # {'pnl': 0, 'bets': 0, 'miss_count': 0} self.daily_pnl_tracker = {} def _calculate_omission_and_momentum(self, history_data): """计算遗漏和动能""" results = [record for record in history_data["result"]] results_array = np.array(results) if len(results_array) < 500: return None, None recent_400 = results_array[-400:] recent_500 = results_array[-500:] omissions = np.full((10, 10), 400) # 计算遗漏 for pos in range(10): for num in range(1, 11): found = np.where(recent_400[:, pos] == num)[0] if len(found) > 0: omissions[pos, num - 1] = len(recent_400) - 1 - found[-1] else: omissions[pos, num - 1] = 400 # 计算动能 (最近 500 期出现次数) momentum = np.zeros((10, 10)) for pos in range(10): for num in range(1, 11): momentum[pos, num - 1] = np.sum(recent_500[:, pos] == num) return omissions, momentum def _get_dynamic_bet_nums(self, pos, omissions, momentum): """根据信号强度动态决定投注注数和号码""" scores = [] for n in range(1, 11): o = omissions[pos, n - 1] m = momentum[pos, n - 1] if o >= self.OMISSIONS_LEVEL_1 and m >= self.MOMENTUM_LEVEL_1: level = 3 # S elif o >= self.OMISSIONS_LEVEL_2 and m >= self.MOMENTUM_LEVEL_2: level = 2 # A elif o >= self.OMISSIONS_LEVEL_3 and m >= self.MOMENTUM_LEVEL_3: level = 1 # B else: level = 0 if level > 0: scores.append({'num': n, 'level': level, 'o': o, 'm': m}) if not scores: return 0, [] max_level = max(s['level'] for s in scores) if max_level == 3: max_bets = 8 elif max_level == 2: max_bets = 6 else: max_bets = 4 scores.sort(key=lambda x: (x['level'], x['o'], x['m']), reverse=True) top_nums = [s['num'] for s in scores[:max_bets]] return len(top_nums), top_nums def _check_position_elite(self, pos, history_data): """检查位置是否为精英位置 (在 predict 中简化为 True)""" return True def check_init_daily_pnl_tracker(self, current_date): _date_range = None for date_range in self.daily_pnl_tracker: start_time, end_time = date_range if start_time <= current_date <= end_time: _date_range = (start_time, end_time) if not _date_range: current_date_start_time = pd.to_datetime(current_date.date()) + timedelta(hours=7, minutes=5) current_date_end_time = current_date_start_time + timedelta(hours=22, minutes=55) _date_range = (current_date_start_time, current_date_end_time) self.daily_pnl_tracker[_date_range] = {'pnl': 0, 'bets': 0, 'miss_count': 0} return _date_range def _check_daily_stop_loss(self, current_date): """检查是否触发日内止损""" stats = self.daily_pnl_tracker[self.check_init_daily_pnl_tracker(current_date)] if not stats: raise Exception(f"{current_date} not in {list(self.daily_pnl_tracker.keys())}") # 先不算水钱 net_pnl = stats['pnl'] # net_pnl = stats['pnl'] + stats['bets'] * self.REBATE_RATE if stats['miss_count'] >= 2: # 连续 miss 取 小值 # print(f"行情不好, 少投点: {stats['miss_count']}") self._bet_this_round = self._bet_this_round if self._bet_this_round < (self.BASE_BET_UNIT * 0.1) else self.BASE_BET_UNIT * 0.1 if net_pnl < -(self.INITIAL_CAPITAL * self.DAILY_STOP_LOSS_RATE): # print("止损 =========") return True return False def update_result(self, period_record, bets_made): """ 在实际开奖后调用，用于更新日内盈亏追踪器 :param period_record: 包含 'result' 和 'time' 的记录 :param bets_made: 实际投注建议 {position: [num1, num2, ...]} """ if not bets_made: return current_date = period_record["time"].iloc[0] # current_date = period_record["time"].dt.strftime("%Y-%m-%d").iloc[0] # current_date = datetime.strptime(period_record['time'], '%Y-%m-%d %H:%M:%S').date() pnl_this_period = 0 bets_this_period = 0 for pos, nums in bets_made["result"].items(): predict_nums = [] bets = 0 for predict_num, bet in nums.items(): if bet: predict_nums.append(predict_num) bets += bet hit = list(period_record['result'])[0][pos] in predict_nums pnl = (self._bet_this_round * self.ODDS if hit else 0) - bets pnl_this_period += pnl bets_this_period += bets date_range = self.check_init_daily_pnl_tracker(current_date) self.daily_pnl_tracker[date_range]['pnl'] += pnl_this_period self.daily_pnl_tracker[date_range]['bets'] += bets_this_period if pnl_this_period < bets_this_period: self.daily_pnl_tracker[date_range]['miss_count'] += 1 else: self.daily_pnl_tracker[date_range]['miss_count'] = 0 def _get_current_actual_pnl(self, current_date): day_data = self.daily_pnl_tracker[self.check_init_daily_pnl_tracker(current_date)] return day_data['pnl'] # return day_data['pnl'] + (day_data['bets'] * self.REBATE_RATE) def _set_bet_level(self, actual_pnl): if actual_pnl == 0: return self.BASE_BET_UNIT * 0.1 if actual_pnl < 0: if actual_pnl < -(self.SAFE_BET_LEVEL_4 * self.INITIAL_CAPITAL): return self.BASE_BET_UNIT * 0.6 if actual_pnl < -(self.SAFE_BET_LEVEL_3 * self.INITIAL_CAPITAL): return self.BASE_BET_UNIT * 0.4 if actual_pnl < -(self.SAFE_BET_LEVEL_2 * self.INITIAL_CAPITAL): return self.BASE_BET_UNIT * 0.3 if actual_pnl < -(self.SAFE_BET_LEVEL_1 * self.INITIAL_CAPITAL): return self.BASE_BET_UNIT * 0.2 return self.BASE_BET_UNIT else: if actual_pnl > self.RADICAL_BET_LEVEL_2 * self.INITIAL_CAPITAL: return self.BASE_BET_UNIT * 1.2 if actual_pnl > self.RADICAL_BET_LEVEL_1 * self.INITIAL_CAPITAL: return self.BASE_BET_UNIT * 1.5 if actual_pnl > self.RADICAL_BET_LEVEL_0 * self.INITIAL_CAPITAL: return self.BASE_BET_UNIT * 2 return self.BASE_BET_UNIT def predict(self, current_date, history_data): """ 核心预测方法 :param current_period_record: 当前期的记录 (用于获取时间) :param history_data: 历史数据列表 :return: 投注建议字典 {position: [num1, num2, ...]} """ # 1. 动态金额调节 actual_pnl = self._get_current_actual_pnl(current_date) self._bet_this_round = self._set_bet_level(actual_pnl) # 2. 日内止损检查 if self._check_daily_stop_loss(current_date): return {} omissions, momentum = self._calculate_omission_and_momentum(history_data) if omissions is None: return {} bet_suggestions = {} for pos in range(10): # 2. 动态位置筛选 # elite_pos = self._get_elite_positions(history_df) # if not elite_pos: # continue # 3. 动态注数投注 num_bets, top_nums = self._get_dynamic_bet_nums(pos, omissions, momentum) if num_bets > 0: bet_suggestions[pos] = {num: self._bet_this_round for num in top_nums} # if self._check_daily_stop_loss(current_date): # for pos, num_bets in bet_suggestions.items(): # bet_suggestions[pos] = { # num: self.BASE_BET_UNIT * self.RADICAL_BET_LEVEL_0 for num, _ in num_bets.items() # } return bet_suggestions