"""
|
低吸策略
|
"""
|
import json
|
import os
|
|
import constant
|
from strategy import strategy_variable
|
|
|
class BackTest:
|
"""
|
回退测试
|
"""
|
|
|
class LowSuctionOriginDataExportManager:
|
"""
|
原始数据导出
|
"""
|
|
def __init__(self, day):
|
"""
|
初始化设置
|
@param day: 日期
|
"""
|
self.day = day
|
|
def __export_logs(self, path_):
|
"""
|
导出日志
|
@param path_:日志路径
|
@return:日志列表
|
"""
|
path_str = f"{constant.get_path_prefix()}/{path_}"
|
if os.path.exists(path_str):
|
with open(path_str, encoding='utf-8', mode='r') as f:
|
return f.readlines()
|
return None
|
|
def __get_log_time(self, line, with_ms=True):
|
if with_ms:
|
time_ = line.split("|")[0].split(" ")[1]
|
else:
|
time_ = line.split("|")[0].split(" ")[1].split(".")[0]
|
return time_
|
|
def __get_async_log_time(self, line, with_ms=True):
|
line = line.split(" - ")[1]
|
if with_ms:
|
time_str = line[line.find("[") + 1:line.find("[") + 9 + 4]
|
else:
|
time_str = line[line.find("[") + 1:line.find("[") + 9]
|
return time_str
|
|
def export_new_block(self):
|
"""
|
导出新板块
|
@return:[(日期,代码,新题材板块)]
|
"""
|
fdatas = []
|
lines = self.__export_logs(f"logs/gp/trade/trade_record.{self.day}.log")
|
for line in lines:
|
time_str = self.__get_async_log_time(line)
|
data = line[line.find("]") + 1:].strip()
|
data = json.loads(data)
|
if data["type"] == 'action' and data["data"].get("type") == '新题材':
|
code = data["code"]
|
blocks = eval(data["data"].get("msg").replace("新题材辨识度设置：", ""))
|
fdatas.append((time_str, code, blocks))
|
fdatas.sort(key=lambda x: x[0])
|
return fdatas
|
|
def export_special_codes(self):
|
"""
|
导出新板块
|
@return:{板块：{代码}}
|
"""
|
lines = self.__export_logs(f"logs/gp/plates/special_codes.{self.day}.log")
|
if lines:
|
line = lines[0]
|
line = line[line.find(" - ") + 3:]
|
return eval(line)
|
return None
|
|
def export_latest_gaobiao(self):
|
"""
|
导出最近高标
|
@return:{板块：{代码}}
|
"""
|
lines = self.__export_logs(f"logs/gp/kpl/kpl_latest_gaobiao.{self.day}.log")
|
if lines:
|
line = lines[0]
|
line = line[line.find(" - ") + 3:]
|
return eval(line)
|
return None
|
|
def export_block_in_datas(self):
|
"""
|
板块流入
|
@return:[(板块,流入金额)]
|
"""
|
lines = self.__export_logs(f"logs/gp/kpl/kpl_jx_in.{self.day}.log")
|
fdatas = []
|
for line in lines:
|
if line.find("原数据：") < 0:
|
continue
|
time_str = self.__get_async_log_time(line)
|
line = line[line.find("原数据：") + 4:]
|
line = line[:line.find("板块：")]
|
temp_list = eval(line)
|
data = [(x[1], x[3]) for x in temp_list]
|
fdatas.append((time_str, data))
|
return fdatas
|
|
def export_block_out_datas(self):
|
"""
|
板块流入
|
@return:[(板块,流入金额)]
|
"""
|
lines = self.__export_logs(f"logs/gp/kpl/kpl_jx_out.{self.day}.log")
|
fdatas = []
|
for line in lines:
|
if line.find("原数据：") < 0:
|
continue
|
time_str = self.__get_async_log_time(line)
|
line = line[line.find("原数据：") + 4:]
|
line = line[:line.find("板块：")]
|
temp_list = eval(line)
|
data = [(x[1], x[3]) for x in temp_list]
|
fdatas.append((time_str, data))
|
return fdatas
|
|
def export_limit_up_list(self):
|
"""
|
涨停列表
|
@return:
|
"""
|
lines = self.__export_logs(f"logs/gp/kpl/kpl_limit_up.{self.day}.log")
|
fdatas = []
|
for line in lines:
|
time_str = self.__get_log_time(line)
|
line = line[line.find(" - ") + 3:]
|
temp_list = eval(line)
|
fdatas.append((time_str, temp_list))
|
return fdatas
|
|
def export_kpl_market_strong(self):
|
"""
|
市场强度
|
@return:
|
"""
|
lines = self.__export_logs(f"logs/gp/kpl/kpl_market_strong.{self.day}.log")
|
fdatas = []
|
for line in lines:
|
time_str = self.__get_log_time(line)
|
line = line[line.find(" - ") + 3:].strip()
|
strong = int(line)
|
fdatas.append((time_str, strong))
|
return fdatas
|
|
def export_zylt_volume(self):
|
"""
|
导出自由流通量
|
@return:
|
"""
|
lines = self.__export_logs(f"logs/gp/code_attribute/zylt_gb.{self.day}.log")
|
for line in lines:
|
line = line[line.find(" - ") + 3:].strip()
|
return eval(line)
|
return None
|
|
def export_big_order_deal(self, min_money=299e4):
|
"""
|
大单成交
|
@return: {"代码":[(买单号, 量, 金额, 时间, 最终成交价)]}
|
"""
|
|
fdatas = {}
|
lines = self.__export_logs(f"logs/huaxin_local/l2/transaction_accurate_big_order.{self.day}.log")
|
if lines:
|
for line in lines:
|
line = line[line.find(" - ") + 3:].strip()
|
data = eval(line)
|
if data[1] != 0:
|
continue
|
if data[2][2] < min_money:
|
continue
|
if data[0] not in fdatas:
|
fdatas[data[0]] = []
|
fdatas[data[0]].append(data[2])
|
return fdatas
|
|
def export_big_sell_order_deal(self, min_money=299e4):
|
"""
|
大单成交
|
@return: {"代码":[(买单号, 量, 金额, 时间, 最终成交价)]}
|
"""
|
fdatas = {}
|
lines = self.__export_logs(f"logs/huaxin_local/l2/transaction_accurate_big_order.{self.day}.log")
|
for line in lines:
|
line = line[line.find(" - ") + 3:].strip()
|
data = eval(line)
|
if data[1] != 1:
|
continue
|
if data[2][2] < min_money:
|
continue
|
if data[0] not in fdatas:
|
fdatas[data[0]] = []
|
fdatas[data[0]].append(data[2])
|
return fdatas
|
|
def export_big_order_deal_by(self, min_money=299e4):
|
"""
|
大单成交
|
@return: {"代码":[(买单号, 量, 金额, 时间, 最终成交价)]}
|
"""
|
buy_order_nos = set()
|
fdatas = {}
|
lines = self.__export_logs(f"logs/huaxin_local/l2/transaction_big_order.{self.day}.log")
|
if lines:
|
for line in lines:
|
line = line[line.find(" - ") + 3:].strip()
|
data = eval(line)
|
if data[1] != 0:
|
continue
|
if data[2][2] < min_money:
|
continue
|
if data[2][0] in buy_order_nos:
|
continue
|
if data[0] not in fdatas:
|
fdatas[data[0]] = []
|
fdatas[data[0]].append(data[2])
|
return fdatas
|
|
def export_big_sell_order_deal_by(self, min_money=299e4):
|
"""
|
大单成交
|
@return: {"代码":[(买单号, 量, 金额, 时间, 最终成交价)]}
|
"""
|
buy_order_nos = set()
|
fdatas = {}
|
lines = self.__export_logs(f"logs/huaxin_local/l2/transaction_big_order.{self.day}.log")
|
for line in lines:
|
line = line[line.find(" - ") + 3:].strip()
|
data = eval(line)
|
if data[1] != 1:
|
continue
|
if data[2][2] < min_money:
|
continue
|
if data[2][0] in buy_order_nos:
|
continue
|
if data[0] not in fdatas:
|
fdatas[data[0]] = []
|
fdatas[data[0]].append(data[2])
|
return fdatas
|
|
def export_code_plates(self):
|
"""
|
代码板块
|
@return: {"代码":[(买单号, 量, 金额, 时间, 最终成交价)]}
|
"""
|
lines = self.__export_logs(f"logs/gp/plates/code_plates.{self.day}.log")
|
for line in lines:
|
if line:
|
line = line[line.find(" - ") + 3:].strip()
|
data = eval(line)
|
return data
|
|
|
class LowSuctionDataManager:
|
"""
|
低吸数据管理
|
"""
|
|
def __init__(self):
|
pass
|
|
def set_tick_data(self, tick_data):
|
"""
|
tick数据: (代码, 现价, 涨幅, 量, 当前时间, 买1价, 买1量, 买2价, 买2量, 更新时间)
|
根据tick数据驱动是否可以买
|
@param tick_data:
|
@return:
|
"""
|
pass
|
|
def set_kpl_limit_up_list(self, limit_up_list):
|
"""
|
设置涨停列表
|
@return:
|
"""
|
|
def set_new_block(self, code, blocks):
|
"""
|
设置新题材
|
@param code: 代码
|
@param blocks: 板块
|
@param is_limit_up:
|
@return:
|
"""
|
|
def set_kpl_market_strong(self, strong):
|
"""
|
开盘啦市场强度
|
@param strong:
|
@return:
|
"""
|
|
def set_code_special_blocks_dict(self, data):
|
"""
|
设置代码的辨识度板块
|
@param data:
|
@return:
|
"""
|
|
def set_code_zylt_gb_dict(self, data):
|
"""
|
设置代码的自由流通股本
|
@param data:
|
@return:
|
"""
|
|
def set_block_in_info(self, data):
|
"""
|
设置板块流入信息
|
@param data:
|
@return:
|
"""
|
pass
|
|
def set_forbidden_refer_codes(self, data):
|
"""
|
设置参考禁止的代码
|
@return:
|
"""
|
pass
|
|
def set_k_bars_dict(self, data):
|
"""
|
设置K线字典
|
@param data:
|
@return:
|
"""
|
|
|
if __name__ == "__main__":
|
fdatas = LowSuctionOriginDataExportManager("2025-04-30").export_code_plates()
|
print(fdatas)
|
result = eval("1>2")
|
print(type(result))
|
print(eval(strategy_variable.大单表达式))
|
