"""
|
策略参数设置
|
"""
|
import json
|
|
from db.mysql_data_delegate import Mysqldb
|
from utils import tool
|
|
|
class StrategyParamsSettings:
|
# 禁止买入
|
STATE_FORBIDDEN_BUY = 0
|
# 正常买入
|
STATE_NORMAL_BUY = 1
|
|
def __init__(self):
|
# 交易状态
|
self.trade_state = 1
|
# 买入金额
|
self.buy_money = 2000
|
# 最大买入票的数量
|
self.max_buy_codes_count = 100
|
# 价格区间
|
self.price_range = (3, 60)
|
# 老题材涨停数
|
self.limit_up_count_of_old_plate = 3
|
# 新题材涨停数
|
self.limit_up_count_of_new_plate = 3
|
# 昨日不能涨停
|
self.cant_yesterday_limit_up = True
|
# 昨日不能跌停
|
self.cant_yesterday_limit_down = True
|
# 昨日不能炸板
|
self.cant_yesterday_open_limit_up = False
|
# 有涨停的交易日数量
|
self.has_limit_up_days = 10
|
# xx个交易日内不能有xx次涨停
|
self.trade_days_count_of_limit_up = 5
|
# 涨停次数
|
self.count_of_limit_up = 3
|
|
# xx个交易日内不能有xx次涨停
|
self.trade_days_count_of_limit_down = 5
|
# 跌停次数
|
self.count_of_limit_down = 3
|
|
# xx个交易日内不能有xx次炸板
|
self.trade_days_count_of_open_limit_up = 5
|
# 炸板次数
|
self.count_of_open_limit_up = 3
|
# 是否可买创业板
|
self.can_buy_ge_code = True
|
# 自由市值范围(单位为亿)
|
self.zyltgb_range = (10, 300)
|
# 是否可买今日涨停过的票
|
self.can_buy_limited_up = False
|
# 最低开盘涨幅
|
self.min_open_rate = -3
|
# 可买的涨幅比例
|
self.avaiable_rates = (-0.03, 0.07)
|
# 今日涨停价需突破XX日最高价,None表示此条数据不生效
|
self.trade_days_count_of_limit_up_price_over_high = None
|
# 今日买价与最高价的差值在XX比例以内
|
self.min_rate_of_highest_and_price = 0.04
|
# 最多高于均价xx比例
|
self.max_rate_than_average_price = 0.03
|
|
def to_json_str(self):
|
d = self.__dict__
|
return json.dumps(d)
|
|
@classmethod
|
def to_obj(cls, json_str):
|
result = json.loads(json_str)
|
obj = StrategyParamsSettings()
|
for k in result:
|
setattr(obj, k, result[k])
|
return obj
|
|
|
@tool.singleton
|
class StrategyParamsSettingsManager:
|
def __init__(self):
|
self.musql = Mysqldb()
|
self.__settings = StrategyParamsSettings()
|
self.__load_data()
|
|
def __load_data(self):
|
sql = f"select `value` from config where `key` = 'low_suction_settings'"
|
setting_str_list = self.musql.select_one(sql)
|
if setting_str_list and setting_str_list[0]:
|
self.__settings = StrategyParamsSettings.to_obj(setting_str_list[0])
|
|
def set_settings(self, settings: StrategyParamsSettings):
|
sql = f"update config set `value` = '{settings.to_json_str()}' where `key` = 'low_suction_settings'"
|
self.musql.execute(sql)
|
self.__settings = settings
|
|
def get_settings(self):
|
return self.__settings
|
|
|
if __name__ == "__main__":
|
settings = StrategyParamsSettingsManager().get_settings()
|
StrategyParamsSettingsManager().set_settings(settings)
|
