279 lines
10 KiB
Python
279 lines
10 KiB
Python
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
|