"""
|
策略变量工厂类
|
"""
|
import datetime
|
import json
|
import os
|
|
from code_attribute import global_data_loader
|
from db import mysql_data_delegate
|
from strategy.data_analyzer import KTickLineAnalyzer, KPLLimitUpDataAnalyzer, K60SLineAnalyzer
|
from strategy.strategy_variable import StockVariables
|
from third_data.history_k_data_manager import HistoryKDataManager
|
from third_data.history_k_data_util import JueJinLocalApi, HistoryKDatasUtils
|
from utils import global_util, tool
|
|
|
class DataLoader:
|
"""
|
数据加载器类，用于集中管理策略变量所需的各类数据加载逻辑
|
"""
|
|
def __init__(self, now_day, cache_path="D:/datas"):
|
"""
|
初始化数据加载器
|
:param now_day: 当前日期，格式为"2025-01-01"
|
"""
|
self.now_day = now_day
|
self.cache_path = cache_path
|
self.jueJinLocalApi = JueJinLocalApi("41c4f5da-2591-11f0-a9c9-f4b5203f67bf",
|
"018db265fa34e241dd6198b7ca507ee0a82ad029")
|
self.trade_days = self.load_trade_days()
|
|
def load_kline_data(self):
|
"""
|
加载日K线数据
|
:return: 日K线数据
|
"""
|
dir_path = os.path.join(self.cache_path, "k_bars")
|
day = self.trade_days[0]
|
dirs = os.listdir(dir_path)
|
k_bar_code_data_dict = {}
|
for d in dirs:
|
if d.find(day) < 0:
|
continue
|
code = d.split("_")[-1][:6]
|
fpath = os.path.join(dir_path, d)
|
with open(fpath, mode='r', encoding='utf-8') as f:
|
lines = f.readlines()
|
line = lines[0]
|
k_bar_code_data_dict[code] = eval(line)
|
return k_bar_code_data_dict
|
|
def load_kline_data_by_day_and_code(self, day, code):
|
path_str = os.path.join(self.cache_path, "k_bars", f"{day}_{code}.txt")
|
if not os.path.exists(path_str):
|
return None
|
with open(path_str, mode='r', encoding='utf-8') as f:
|
lines = f.readlines()
|
line = lines[0]
|
return eval(line)
|
return None
|
|
def load_minute_data(self):
|
"""
|
加载分钟K线数据
|
:return: 分钟K线数据字典
|
"""
|
dir_path = os.path.join(self.cache_path, "bar_60s")
|
day = self.trade_days[0]
|
dirs = os.listdir(dir_path)
|
k_bar_code_data_dict = {}
|
for d in dirs:
|
if d.find(day) < 0:
|
continue
|
code = d.split("_")[-1][:6]
|
sub_dir_path = os.path.join(dir_path, d)
|
sub_dirs = os.listdir(sub_dir_path)
|
date_datas = {}
|
for sub_dir in sub_dirs:
|
fpath = os.path.join(sub_dir_path, sub_dir)
|
date = sub_dir[:10]
|
with open(fpath, mode='r', encoding='utf-8') as f:
|
lines = f.readlines()
|
line = lines[0]
|
date_datas[date] = eval(line)
|
k_bar_code_data_dict[code] = date_datas
|
return k_bar_code_data_dict
|
|
def load_tick_data(self):
|
"""
|
加载当日的tick数据
|
:return:tick数据字典
|
"""
|
code_ticks_dict = {}
|
tick_dir_path = os.path.join(self.cache_path, "ticks")
|
if os.path.exists(tick_dir_path):
|
fs = os.listdir(tick_dir_path)
|
for f in fs:
|
if f.find(self.now_day) < 0:
|
continue
|
code = f.split("_")[1][:6]
|
tick_path = os.path.join(tick_dir_path, f)
|
with open(tick_path, mode='r') as f:
|
lines = f.readlines()
|
line = lines[0]
|
line = line.replace("datetime.datetime(", "\"datetime.datetime(").replace("('PRC'))", "('PRC'))\"")
|
line = line.replace("'", "\"").replace("\"PRC\"", "'PRC'")
|
ticks = json.loads(line)
|
for t in ticks:
|
if t["created_at"].find("datetime.datetime") >= 0:
|
created_at_line = t["created_at"].replace("datetime.datetime(", "")
|
sts = created_at_line.split(",")
|
year, month, day, hour, minute, second = sts[0].strip(), sts[1].strip(), sts[2].strip(), \
|
sts[
|
3].strip(), \
|
sts[4].strip(), (
|
sts[5].strip() if len(sts) >= 7 else 0)
|
dt = datetime.datetime(int(year), int(month), int(day), int(hour), int(minute), int(second))
|
t["created_at"] = dt.strftime("%Y-%m-%d %H:%M:%S")
|
# 整理为分钟K线
|
code_ticks_dict[code] = ticks
|
return code_ticks_dict
|
|
def load_limit_up_data(self):
|
"""
|
加载涨停数据
|
:return: 涨停数据记录
|
"""
|
mysql = mysql_data_delegate.Mysqldb()
|
results = mysql.select_all(
|
f"select _code, _day, _hot_block_name from kpl_limit_up_record where _day>='{self.trade_days[-1]}' and _day<='{self.trade_days[0]}' and _open=0")
|
return results
|
|
def load_trade_days(self):
|
"""
|
加载交易日列表,now_day前120个交易日
|
:return: 交易日列表
|
"""
|
# 将now_day转为datetime类型
|
current_date = datetime.datetime.strptime(self.now_day, '%Y-%m-%d').date()
|
pre_date = current_date - datetime.timedelta(days=1)
|
one_year_ago = (pre_date - datetime.timedelta(days=365)).strftime('%Y-%m-%d')
|
pre_date = pre_date.strftime('%Y-%m-%d')
|
|
trade_days = self.jueJinLocalApi.get_trading_dates(one_year_ago, pre_date)
|
trade_days.sort(reverse=True)
|
trade_days = trade_days[:120]
|
return trade_days
|
|
def load_next_trade_day(self):
|
"""
|
加载交易日列表,now_day前120个交易日
|
:return: 交易日列表
|
"""
|
next_trade_day = self.jueJinLocalApi.get_next_trading_date(self.now_day)
|
return next_trade_day
|
|
def load_target_codes(self):
|
"""
|
获取特殊的代码，需要订阅300w以上的大单
|
@return: 代码集合, 收盘价字典：{"code":收盘价}
|
"""
|
try:
|
global_data_loader.load_zyltgb_volume_from_db()
|
pre_close_price_dict = {}
|
zylt_volume_map = global_util.zylt_volume_map
|
codes = set()
|
last_trade_day = self.trade_days[0]
|
for code in zylt_volume_map:
|
volume = zylt_volume_map.get(code)
|
# 今日涨停价要突破昨日最高价
|
k_bars = HistoryKDataManager().get_history_bars(code, last_trade_day)
|
if k_bars and 30e8 <= k_bars[0]["close"] * volume * tool.get_limit_up_rate(code) <= 300e8:
|
# 自由流通市值在30亿-300亿以上
|
limit_up_price = round(tool.get_limit_up_rate(code) * k_bars[0]["close"], 2)
|
max_price_info = max(k_bars[:1], key=lambda x: x["high"])
|
if limit_up_price > max_price_info["high"]:
|
# 今日涨停价要突破昨日最高价
|
codes.add(code)
|
pre_close_price_dict[code] = k_bars[0]["close"]
|
return codes, pre_close_price_dict
|
except Exception as e:
|
return set(), None
|
|
|
class StrategyVariableFactory:
|
@staticmethod
|
def create_from_history_data(kline_data_1d, kline_data_60s_dict, limit_up_data_records,
|
trade_days) -> StockVariables:
|
"""
|
根据历史数据创建StockVariables对象
|
:param kline_data_1d: 日K线数据
|
:param kline_data_60s_dict:分钟k线字典：{"2025-01-01":{Bar格式}}
|
:param limit_up_data_records: 代码的历史涨停数据：[(代码,日期,涨停原因)]
|
:param trade_days: 交易日列表，从大到小倒序排列
|
:return: StockVariables实例
|
"""
|
instance = StockVariables()
|
# 设置K线相关属性
|
instance.昨日成交量 = KTickLineAnalyzer.get_yesterday_volume(kline_data_1d)
|
instance.昨日收盘价 = KTickLineAnalyzer.get_yesterday_close(kline_data_1d)
|
instance.昨日最高价 = KTickLineAnalyzer.get_yesterday_high(kline_data_1d)
|
instance.昨日成交额 = KTickLineAnalyzer.get_yesterday_amount(kline_data_1d)
|
instance.昨日非涨停 = not KTickLineAnalyzer.is_yesterday_limit_up(kline_data_1d)
|
instance.昨日非炸板 = not KTickLineAnalyzer.is_yesterday_exploded(kline_data_1d)
|
instance.昨日非跌停 = not KTickLineAnalyzer.is_yesterday_limit_down(kline_data_1d)
|
day_counts = [5, 10, 30, 60, 120]
|
for day in day_counts:
|
instance.__setattr__(f"日最高价_{day}", KTickLineAnalyzer.get_recent_days_high(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日最高量_{day}", KTickLineAnalyzer.get_recent_days_max_volume(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"涨停数_{day}", KTickLineAnalyzer.get_recent_limit_up_count(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"炸板数_{day}", KTickLineAnalyzer.get_recent_exploded_count(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"跌停数_{day}", KTickLineAnalyzer.get_recent_limit_down_count(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日首板溢价率_{day}", KTickLineAnalyzer.get_first_limit_up_avg_premium(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日首板炸板溢价率_{day}",
|
KTickLineAnalyzer.get_first_exploded_avg_premium(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日首板个数_{day}", KTickLineAnalyzer.get_first_limit_up_days(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日二板个数_{day}", KTickLineAnalyzer.get_second_limit_up_days(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日三板个数_{day}", KTickLineAnalyzer.get_third_limit_up_days(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日大等于四板个数_{day}",
|
KTickLineAnalyzer.get_fourth_or_more_limit_up_days(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日首跌停个数_{day}", KTickLineAnalyzer.get_first_limit_down_days(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日2次跌停个数_{day}", KTickLineAnalyzer.get_second_limit_down_days(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日3次跌停个数_{day}", KTickLineAnalyzer.get_third_limit_down_days(kline_data_1d, day))
|
for day in day_counts:
|
instance.__setattr__(f"日大等于4次跌停个数_{day}",
|
KTickLineAnalyzer.get_fourth_or_more_limit_down_days(kline_data_1d, day))
|
|
for day in day_counts:
|
days = trade_days[:day]
|
instance.__setattr__(f"日个股最正的原因_{day}",
|
KPLLimitUpDataAnalyzer.get_most_common_reasons(limit_up_data_records, min_day=days[-1],
|
max_day=days[0]))
|
if kline_data_60s_dict:
|
for day in day_counts:
|
# 获取日K最高量的信息
|
volume, k_data = KTickLineAnalyzer.get_recent_days_max_volume(kline_data_1d, day)
|
d = k_data["bob"][:10]
|
kline_data_60s = kline_data_60s_dict.get(d)
|
fdata = K60SLineAnalyzer.get_close_price_of_max_volume(kline_data_60s)
|
instance.__setattr__(f"日高量分时最高量价_{day}", fdata)
|
kline_data_60s = kline_data_60s_dict.get(trade_days[0])
|
fdata = K60SLineAnalyzer.get_close_price_of_max_volume(kline_data_60s)
|
instance.__setattr__(f"昨日分时最高量价", fdata)
|
return instance
|
|
|
if __name__ == "__main__":
|
# instance = StockVariables()
|
# day = 5
|
# instance.__setattr__(f"日最高价_{day}", 12.00)
|
# print(instance.日最高价_5)
|
DataLoader("2025-05-06").load_tick_data()
|
