策略完善

Administrator

2025-06-12 9399db7dfedc7a36ed1cead1610f1039365bd770

策略完善

 code_attribute/gpcode_manager.py

@@ -655,36 +655,38 @@
    return list


@tool.singleton
class CodePrePriceManager:
    __price_pre_cache = {}
    __redisManager = redis_manager.RedisManager(0)

    def __init__(self):
        # åŠ è½½æ•°æ®
        fdatas = log_export.load_pre_close_price()
        if fdatas:
            for code in fdatas:
                self.__price_pre_cache[code] = round(float(fdatas.get(code)), 2)

    # èŽ·å–收盘价
    @classmethod
    def get_price_pre(cls, code):
    def get_price_pre(self, code):
        fdatas = log_export.load_pre_close_price()
        if code in fdatas:
            return round(float(fdatas.get(code)), 2)
        return None

    # èŽ·å–缓存
    @classmethod
    def get_price_pre_cache(cls, code):
        if code in cls.__price_pre_cache:
            return float(cls.__price_pre_cache[code])
        val = cls.get_price_pre(code)
        if val:
            cls.__price_pre_cache[code] = val
        return val
    def get_price_pre_cache(self, code):
        if code in self.__price_pre_cache:
            return self.__price_pre_cache[code]
        return None

    # è®¾ç½®æ”¶ç›˜ä»·
    @classmethod
    def set_price_pre(cls, code, price, force=False):
        if code in cls.__price_pre_cache and not force:
    def set_price_pre(self, code, price, force=False):
        if code in self.__price_pre_cache and not force:
            return
        price = round(float(price), 2)
        logger_pre_close_price.info(f"{code}-{price}")
        cls.__price_pre_cache[code] = price
        self.__price_pre_cache[code] = price


__limit_up_price_dict = {}
@@ -695,7 +697,7 @@
    # è¯»å–内存中的值
    if code in __limit_up_price_dict:
        return __limit_up_price_dict[code]
    price = CodePrePriceManager.get_price_pre_cache(code)
    price = CodePrePriceManager().get_price_pre_cache(code)
    if price is None:
        return None
    limit_up_price = tool.to_price(decimal.Decimal(str(price)) * decimal.Decimal(tool.get_limit_up_rate(code)))
@@ -730,7 +732,7 @@

# èŽ·å–跌停价
def get_limit_down_price(code):
    price = CodePrePriceManager.get_price_pre_cache(code)
    price = CodePrePriceManager().get_price_pre_cache(code)
    if price is None:
        return None
    return tool.to_price(decimal.Decimal(str(price)) * decimal.Decimal(f"{tool.get_limit_down_rate(code)}"))

 main.py

@@ -7,9 +7,12 @@

from api import outside_api_callback
from api.outside_api_command_manager import ApiCommandManager
from db.redis_manager_delegate import RedisUtils
from huaxin_client import l2_market_client, trade_client
from log_module import async_log_util
from log_module.log import logger_debug
from server import data_server
from strategy import strategy_manager
from strategy.env_info import RealTimeEnvInfo
from third_data import hx_qc_value_util
from trade.huaxin import huaxin_trade_api
@@ -24,8 +27,10 @@
                if data.get("type") == 'set_target_codes':
                    # [(代码, æ—¶é—´æˆ³, ä»·æ ¼, æ€»äº¤æ˜“量, æ€»äº¤æ˜“额, ä¹°5, å–5)]
                    market_data_list = data["data"]["data"]
                    strategy_manager.low_suction_strtegy.add_ticks(market_data_list)
                    RealTimeEnvInfo().ticks = (tool.get_now_time_str(), len(market_data_list))
            except:
            except Exception as e:
                logger_debug.exception(e)
                time.sleep(0.1)

    queue_l1_w_strategy_r: multiprocessing.Queue = multiprocessing.Queue()
@@ -50,6 +55,12 @@
    # -----启动data_server-----
    threading.Thread(target=lambda: data_server.run("127.0.0.1", 9008), daemon=True).start()

    # å¯åŠ¨æœ¬åœ°æ—¥å¿—
    threading.Thread(target=lambda: async_log_util.run_sync(), daemon=True).start()

    # å¯åŠ¨redis同步
    threading.Thread(target=lambda: RedisUtils.run_loop(), daemon=True).start()

    # --------启动本地API接口----------
    manager = ApiCommandManager(middle_api_protocol.SERVER_HOST, middle_api_protocol.SERVER_PORT, outside_api_callback.MyAPICallback())
    manager.run(blocking=False)

 server/data_server.py

@@ -5,6 +5,7 @@
from http.server import BaseHTTPRequestHandler

from log_module.log import logger_system, logger_debug, logger_request_api
from strategy import strategy_manager

from utils import tool
from log_module import async_log_util
@@ -36,21 +37,27 @@
        path = self.path
        url = urlparse.urlparse(path)
        result_str = ""
        if url.path == "/upload_big_order_datas":
            # æŽ¥æ”¶æˆäº¤å¤§å•æ•°æ®
            params = self.__parse_request()
            logger_debug.info("upload_big_order_datas:{}", f"{params}")
        elif url.path == "/upload_block_in_datas":
            # æŽ¥æ”¶æ¿å—流入数据
            params = self.__parse_request()
            logger_debug.info("upload_block_in_datas:{}", f"{params}")
            result_str = json.dumps({"code": 0})
        elif url.path == "upload_limit_up_list":
            params = self.__parse_request()
            logger_debug.info("upload_limit_up_list:{}", f"{params}")
            result_str = json.dumps({"code": 0})
        else:
            pass
        try:
            if url.path == "/upload_big_order_datas":
                # æŽ¥æ”¶æˆäº¤å¤§å•æ•°æ®
                params = self.__parse_request()
                strategy_manager.low_suction_strtegy.add_big_orders(params)
                # logger_debug.info("upload_big_order_datas:{}", f"{params}")
            elif url.path == "/upload_block_in_datas":
                # æŽ¥æ”¶æ¿å—流入数据
                params = self.__parse_request()
                strategy_manager.low_suction_strtegy.add_block_in(params)
                # logger_debug.info("upload_block_in_datas:{}", f"{params}")
                result_str = json.dumps({"code": 0})
            elif url.path == "/upload_limit_up_list":
                params = self.__parse_request()
                strategy_manager.low_suction_strtegy.add_limit_up_list(params)
                # logger_debug.info("upload_limit_up_list:{}", f"{params}")
                result_str = json.dumps({"code": 0})
            else:
                pass
        except Exception as e:
            logger_debug.exception(e)
        self.__send_response(result_str)

    def __send_response(self, data):

 strategy/strategy_manager.py

New file
@@ -0,0 +1,388 @@
"""
策略管理
"""
import json

from code_attribute import gpcode_manager, code_nature_analyse
from db import redis_manager_delegate as redis_manager
from db.mysql_data_delegate import Mysqldb
from db.redis_manager_delegate import RedisUtils
from strategy.data_analyzer import KPLLimitUpDataAnalyzer
from strategy.low_suction_strategy import LowSuctionOriginDataExportManager
from strategy.strategy_params_settings import StrategyParamsSettingsManager
from strategy.strategy_variable import StockVariables
from strategy.strategy_variable_factory import DataLoader, StrategyVariableFactory
import constant
from third_data import kpl_util
from trade.trade_manager import DealCodesManager
from utils import huaxin_util, tool


@tool.singleton
class TickSummaryDataManager:
    """
    Tick概要数据
    """
    __db = 13

    # æˆäº¤ä»£ç çš„订单信息：{代码:{交易id:(量，价格，系统订单号)}}

    def __init__(self):
        # å¼€ç›˜ä»·ä¿¡æ¯
        self.open_price_info_dict = {}
        # æœ€ä½Žä»·
        self.low_price_dict = {}
        # æœ€é«˜ä»·
        self.high_price_info_dict = {}

        self.musql = Mysqldb()
        self.redis_manager = redis_manager.RedisManager(self.__db)
        self.__load_data()

    def __get_redis(self):
        return self.redis_manager.getRedis()

    def __load_data(self):
        keys = RedisUtils.keys(self.__get_redis(), "tick_open_price_info-*")
        if keys:
            for k in keys:
                code = k.split("-")[1]
                val = RedisUtils.get(self.__get_redis(), k)
                if val:
                    self.open_price_info_dict[code] = json.loads(val)
        keys = RedisUtils.keys(self.__get_redis(), "tick_high_price-*")
        if keys:
            for k in keys:
                code = k.split("-")[1]
                val = RedisUtils.get(self.__get_redis(), k)
                if val:
                    self.high_price_info_dict[code] = json.loads(val)

        keys = RedisUtils.keys(self.__get_redis(), "tick_low_price-*")
        if keys:
            for k in keys:
                code = k.split("-")[1]
                val = RedisUtils.get(self.__get_redis(), k)
                if val:
                    self.low_price_dict[code] = round(float(val))

    def set_open_price_info(self, code, info):
        """
        è®¾ç½®å¼€ç›˜ä¿¡æ¯
        @param code:
        @param info:
        @return:
        """
        self.open_price_info_dict[code] = info
        RedisUtils.setex_async(self.__db, f"tick_open_price_info-{code}", tool.get_expire(), json.dumps(info))

    def set_high_price_info(self, code, info):
        """
        è®¾ç½®æœ€é«˜ä»·
        @param code:
        @param info:(ä»·æ ¼, æ—¶é—´)
        @return:
        """
        self.high_price_info_dict[code] = info
        RedisUtils.setex_async(self.__db, f"tick_high_price-{code}", tool.get_expire(), json.dumps(info))

    def set_low_price(self, code, price):
        """
        è®¾ç½®æœ€ä½Žä»·
        @param code:
        @param price:
        @return:
        """
        self.low_price_dict[code] = price
        RedisUtils.setex_async(self.__db, f"tick_low_price-{code}", tool.get_expire(), price)


class LowSuctionStrategy:
    """
    ä½Žå¸ç­–ç•¥
    """

    def __init__(self, day, script_name="低吸脚本_辨识度_v6.py", settings=StrategyParamsSettingsManager().get_settings()):
        self.now_day = day
        # ä¹°å¤§å•ï¼š{代码:[大单数据]}
        self.big_order_buy = {}
        # å–大单
        self.big_order_sell = {}
        # è‡ªç”±æµé€šé‡
        self.zylt_volume_dict = {}
        # åŽ†å²æ—¥K数据
        self.kline_data = {}
        # åŽ†å²æ¶¨åœæ•°æ®
        self.limit_up_record_data = {}
        # åŽ†å²æ•°æ®
        self.timeline_data = {}
        # ä»Šæ—¥æ•°æ®
        self.current_data = {}
        # ç›®æ ‡ä»£ç çš„板块
        self.code_plates_for_buy = {}
        # ä»£ç çš„板块（常规）
        self.code_plates = {}
        # ä¸‹ä¸€ä¸ªäº¤æ˜“æ—¥
        self.next_trade_day = None
        # ç›®æ ‡ä»£ç 
        self.fcodes = set()
        # å˜é‡å¯¹è±¡
        self.stock_variables_dict = {}
        # å½“前涨停列表
        self.current_limit_up_list = []
        # å½“前涨停的板块所包含的代码
        self.current_limit_up_plate_codes = {}
        # å½“前板块净流入
        self.current_block_in_datas = []

        # åŠ è½½ç­–略脚本文件
        with open(script_name, mode='r', encoding='utf-8') as f:
            lines = f.readlines()
            scripts = "\n".join(lines)
            # æ³¨é‡ŠæŽ‰é‡Œé¢çš„import与变量
            scripts = scripts.replace("from ", "#from ").replace("sv = ", "#sv =  ").replace("settings = ",
                                                                                             "#settings =  ").replace(
                "target_code = ", "#target_code =  ")

            self.scripts = scripts
        self.settings = settings

        self.data_loader = DataLoader(self.now_day, cache_path=f"{constant.get_path_prefix()}/datas")
        self.__LowSuctionOriginDataExportManager = LowSuctionOriginDataExportManager(self.now_day)
        self.load_data()

    def load_data(self):
        # åŠ è½½åŽ†å²æ•°æ®
        self.__load_before_date_data_by_timeline()
        self.__load_current_date_data_by_timeline()
        self.fcodes = set(self.code_plates_for_buy.keys()) & set(self.kline_data.keys())

    def __load_before_date_data_by_timeline(self):
        """
        åŠ è½½å›žæµ‹æ—¥æœŸä¹‹å‰çš„K线数据与历史涨停数据
        :return: æŒ‰æ—¶é—´æŽ’序的数据列表
        """
        trade_days = self.data_loader.trade_days
        # åŠ è½½åŽ†å²æ•°æ®
        self.kline_data = self.data_loader.load_kline_data()
        self.limit_up_record_data = self.data_loader.load_limit_up_data()
        self.next_trade_day = self.data_loader.load_next_trade_day()
        if not trade_days:
            raise Exception("交易日历获取失败")
        if not self.kline_data:
            raise Exception("历史日K获取失败")
        if not self.limit_up_record_data:
            raise Exception("历史涨停获取失败")

    def __load_current_date_data_by_timeline(self):
        """
        åŠ è½½å›žæµ‹æ—¥æœŸå½“天的数据，将这些数据根据秒切片
        :param day: æ—¥æœŸï¼Œæ ¼å¼ä¸º"YYYY-MM-DD
        :return: æŒ‰æ—¶é—´æŽ’序的数据列表
        """
        IS_BY_BIG_ORDER = False
        BIG_ORDER_MONEY_THRESHOLD = 200e4
        big_order_deals = self.__LowSuctionOriginDataExportManager.export_big_order_deal(BIG_ORDER_MONEY_THRESHOLD)
        if not big_order_deals or IS_BY_BIG_ORDER:
            big_order_deals = self.__LowSuctionOriginDataExportManager.export_big_order_deal_by(
                BIG_ORDER_MONEY_THRESHOLD)
        self.big_order_buy = big_order_deals
        # è½¬æ¢æ ¼å¼ä¸ºï¼š{时间: [("代码", (买单号, é‡, é‡‘额, æ—¶é—´, æœ€ç»ˆæˆäº¤ä»·))]

        big_sell_order_deals = self.__LowSuctionOriginDataExportManager.export_big_sell_order_deal(
            BIG_ORDER_MONEY_THRESHOLD)
        if not big_sell_order_deals or IS_BY_BIG_ORDER:
            big_sell_order_deals = self.__LowSuctionOriginDataExportManager.export_big_sell_order_deal_by(
                BIG_ORDER_MONEY_THRESHOLD)
        self.big_order_sell = big_sell_order_deals
        # åŠ è½½è‡ªç”±æµé€šé‡
        self.zylt_volume_dict = self.__LowSuctionOriginDataExportManager.export_zylt_volume()

        # åŠ è½½æ¿å—代码
        code_plates_dict = self.__LowSuctionOriginDataExportManager.export_code_plates()

        plate_codes = self.data_loader.load_target_plate_and_codes()
        code_plates_dict_for_buy = {}
        for p in plate_codes:
            for code in plate_codes.get(p):
                if code not in code_plates_dict_for_buy:
                    code_plates_dict_for_buy[code] = set()
                code_plates_dict_for_buy[code].add(p)
        self.code_plates_for_buy = code_plates_dict_for_buy
        self.code_plates = code_plates_dict

        if not self.zylt_volume_dict:
            raise Exception("无自由流通数据")
        if not self.code_plates:
            raise Exception("无板块数据")
        if not self.code_plates_for_buy:
            raise Exception("无目标票的买入数据")

    def __init_stock_variables(self, code_):
        """
        åˆå§‹åŒ–变量
        @param code_:
        @return:
        """
        if code_ in self.stock_variables_dict:
            return

        stock_variables = StrategyVariableFactory.create_from_history_data(
            self.kline_data.get(code_), None,
            self.limit_up_record_data.get(code_), self.data_loader.trade_days)

        # åŠ è½½ä»Šæ—¥æ¶¨åœä»·
        pre_close = self.kline_data.get(code_)[0]["close"]
        stock_variables.今日涨停价 = round(float(gpcode_manager.get_limit_up_price_by_preprice(code_, pre_close)), 2)
        stock_variables.自由流通市值 = self.zylt_volume_dict.get(code_) * pre_close
        # èŽ·å–代码板块
        stock_variables.代码板块 = self.code_plates_for_buy.get(code_)
        is_price_too_high = code_nature_analyse.is_price_too_high_in_days(code_, self.kline_data.get(code_),
                                                                          stock_variables.今日涨停价)
        stock_variables.六个交易日涨幅过高 = is_price_too_high[0]
        stock_variables.辨识度代码 = self.fcodes
        for day in [2, 5, 10, 30, 60, 120]:
            days = self.data_loader.trade_days[:day]
            stock_variables.__setattr__(f"日出现的板块_{day}",
                                        KPLLimitUpDataAnalyzer.get_limit_up_reasons(
                                            self.limit_up_record_data, min_day=days[-1],
                                            max_day=days[0]))
        stock_variables.连续老题材 = KPLLimitUpDataAnalyzer.get_continuous_limit_up_reasons(
            self.limit_up_record_data, self.data_loader.trade_days[:2])

        # åŠ è½½Tick信息
        open_price_info = TickSummaryDataManager().open_price_info_dict.get(code_)
        if open_price_info:
            stock_variables.今日开盘价 = open_price_info[0]
            stock_variables.今日开盘涨幅 = open_price_info[1]
        high_price_info = TickSummaryDataManager().high_price_info_dict.get(code_)
        if high_price_info:
            stock_variables.今日最高价信息 = high_price_info

        low_price = TickSummaryDataManager().low_price_dict.get(code_)
        if low_price:
            stock_variables.今日最低价 = low_price
        self.stock_variables_dict[code_] = stock_variables

    def add_big_orders(self, big_orders):
        """
        æ·»åŠ å¤§å•
        @param big_orders: [(代码, ä¹°/卖, [订单号,量,金额,最后时间戳,最后价格, åˆå§‹æ—¶é—´æˆ³, åˆå§‹ä»·æ ¼])] å¦‚：[ ('002741', 0, [475820, 91600, 1610328, 92500000, 17.58, 92500000, 17.58])]
        @return:
        """
        for d in big_orders:
            code = d[0]
            if d[1] == 0:
                # ä¹°å•
                if code not in self.big_order_buy:
                    self.big_order_buy[code] = []
                self.big_order_buy[code].append(d[2])
            else:
                # å–单
                if code not in self.big_order_sell:
                    self.big_order_sell[code] = []
                self.big_order_sell[code].append(d[2])
        # é©±åŠ¨ä¸‹å•

    def add_ticks(self, ticks):
        """
        æ·»åŠ tick数据
        @param ticks:
        @return:
        """
        fticks = [tick for tick in ticks if tick[0] in self.fcodes]
        # åˆå§‹åŒ–对象
        for tick in fticks:
            code = tick[0]
            self.__init_stock_variables(code)
            stock_variables = self.stock_variables_dict.get(code)

            # (代码, æ—¶é—´æˆ³, ä»·æ ¼, æ€»äº¤æ˜“量, æ€»äº¤æ˜“额, ä¹°5, å–5)
            code, time_str, price, cum_volume, cum_amount, buy_5 = tick[0], huaxin_util.convert_time(tick[1]), tick[2], \
                                                                   tick[
                                                                       3], tick[4], tick[5]

            # è®¡ç®—tick数据的值
            if time_str < '09:30:00':
                if price > 0:
                    stock_variables.今日开盘价 = price
                else:
                    stock_variables.今日开盘价 = buy_5[0][0]
                    # ä»Šæ—¥å¼€ç›˜æ¶¨å¹…
                stock_variables.今日开盘涨幅 = round((stock_variables.今日开盘价 - stock_variables.昨日收盘价) / stock_variables.昨日收盘价,
                                               4)
                TickSummaryDataManager().set_open_price_info(code, (stock_variables.今日开盘价, stock_variables.今日开盘涨幅))
            stock_variables.今日成交量 = cum_volume
            stock_variables.今日成交额 = cum_amount
            stock_variables.当前价 = price
            if not stock_variables.今日最高价信息 or price > stock_variables.今日最高价信息[0]:
                stock_variables.今日最高价信息 = (price, time_str)
                TickSummaryDataManager().set_high_price_info(code, stock_variables.今日最高价信息)
            if not stock_variables.今日最低价 or price < stock_variables.今日最低价:
                stock_variables.今日最低价 = price
                TickSummaryDataManager().set_low_price(code, stock_variables.今日最低价)

    def add_limit_up_list(self, limit_up_list):
        """
        æ¶¨åœåˆ—表数据
        @param limit_up_list:
        @return:
        """
        self.current_limit_up_list = limit_up_list
        current_limit_up_list = [x for x in limit_up_list if kpl_util.get_high_level_count(x[4]) < 3]
        most_real_kpl_plate_limit_up_codes_info = {}
        current_limit_up_dict = {x[0]: x for x in current_limit_up_list}
        codes = set(current_limit_up_dict.keys())
        for code in codes:
            plates = self.code_plates.get(code)
            if not plates:
                plates = {current_limit_up_dict.get(code)[5]}
            plates -= constant.KPL_INVALID_BLOCKS
            if plates:
                for p in plates:
                    if p not in most_real_kpl_plate_limit_up_codes_info:
                        most_real_kpl_plate_limit_up_codes_info[p] = []
                    most_real_kpl_plate_limit_up_codes_info[p].append(code)
        self.current_limit_up_plate_codes = most_real_kpl_plate_limit_up_codes_info

    def add_block_in(self, block_in_datas):
        """
        æ·»åŠ æ¿å—净流入
        @param block_in_datas:
        @return:
        """
        blocks = [x[0] for x in block_in_datas if x[1] > 0]
        _block_in_datas = blocks[:20]
        self.current_block_in_datas = _block_in_datas

    def __run(self, code, sv: StockVariables):
        # è¿è¡Œä»£ç 
        # æ³¨å…¥å¤§å•
        sv.今日大单数据 = self.big_order_buy.get(code)
        sv.今日卖大单数据 = self.big_order_sell.get(code)
        # æ³¨å…¥æ¿å—涨停代码
        sv.开盘啦最正板块涨停 = self.current_limit_up_plate_codes
        # æ³¨å…¥æ¿å—流入信息
        if self.current_block_in_datas:
            sv.资金流入板块 = self.current_block_in_datas
        # æ³¨å…¥å·²æˆäº¤ä»£ç 
        place_order_plate_codes = DealCodesManager().get_place_order_plate_codes()
        sv.板块成交代码 = place_order_plate_codes
        sv.成交代码 = DealCodesManager().get_deal_codes()
        global_dict = {
            "sv": sv,
            "target_code": code,
            "settings": self.settings
        }
        exec(self.scripts, global_dict)
        compute_result = global_dict["compute_result"]
        if compute_result[0]:
            if code in sv.成交代码:
                return
            # å¯ä»¥ä¸‹å•
            # åˆ¤æ–­æ˜¯å¦å¯ä»¥ä¹°
            for b in compute_result[3]:
                DealCodesManager().place_order(b, code)

# å½“前的低吸策略对象
low_suction_strtegy = LowSuctionStrategy(tool.get_now_date_str())

 strategy/strategy_params_settings.py

@@ -19,7 +19,7 @@
        # ä¹°å…¥é‡‘额
        self.buy_money = 2000
        # æœ€å¤§ä¹°å…¥ç¥¨çš„数量
        self.max_buy_codes_count = 10
        self.max_buy_codes_count = 100
        # ä»·æ ¼åŒºé—´
        self.price_range = (3, 60)
        # è€é¢˜ææ¶¨åœæ•°
@@ -50,12 +50,12 @@
        self.count_of_open_limit_up = 3
        # æ˜¯å¦å¯ä¹°åˆ›ä¸šæ¿
        self.can_buy_ge_code = True
        # è‡ªç”±å¸‚值范围
        self.zyltgb_range = (10e8, 300e8)
        # è‡ªç”±å¸‚值范围(单位为亿)
        self.zyltgb_range = (10, 300)
        # æ˜¯å¦å¯ä¹°ä»Šæ—¥æ¶¨åœè¿‡çš„票
        self.can_buy_limited_up = False
        # æœ€ä½Žå¼€ç›˜æ¶¨å¹…
        self.min_open_rate = 0.00001
        self.min_open_rate = -3
        # å¯ä¹°çš„涨幅比例
        self.avaiable_rates = (-0.03, 0.07)
        # ä»Šæ—¥æ¶¨åœä»·éœ€çªç ´XX日最高价,None表示此条数据不生效

 strategy/strategy_variable.py

@@ -201,10 +201,11 @@
        # å½“日板块：{"板块1","板块2"}
        self.代码板块 = None
        self.新代码板块 = None
        self.代码精选板块=None
        self.代码精选板块 = None

        # æ¿å—成交代码：{"板块":{"代码1","代码2"}}
        self.板块成交代码 = {}
        self.成交代码 = set()

        # æµå…¥æ¿å—
        self.资金流入板块 = []

 strategy/strategy_variable_factory.py

@@ -235,10 +235,10 @@
                results = [x for x in results if
                           (tool.is_can_buy_code(x[0]) and x[0] in valid_codes and x[0] not in exclude_codes)]
                # å–前1/3且涨停数是前10
                # max_count = len(results) // 3 if len(results) % 3 == 0 else len(results) // 3 + 1
                # results = results[:max_count]
                # # å–前10
                # results = results[:10]
                max_count = len(results) // 2 if len(results) % 2 == 0 else len(results) // 2 + 1
                results = results[:max_count]
                # å–前10
                results = results[:10]
                codes = [x[0] for x in results]
                fresults[kpl_util.filter_block(b)] = codes
        return fresults
@@ -637,7 +637,7 @@


if __name__ == "__main__":
    __DataLoader = DataLoader("2025-06-09")
    __DataLoader = DataLoader("2025-06-12")
    # __test_jx_blocks(__DataLoader)

    # instance = StockVariables()
@@ -653,7 +653,9 @@
    # print(result_dict["301279"])

    results = __DataLoader.load_target_plate_and_codes()
    plates = ["医药"]
    # for k in results:
    #     print(k, results[k])
    plates = ["汽车零部件", "稀土永磁", "化工", "医药", "光伏"]
    print("==========新题材=======")
    for p in plates:
        print(p, results.get(p))

 strategy/test.py

@@ -1,3 +1,4 @@
from strategy import strategy_manager
from strategy.strategy_variable import StockVariables


@@ -18,8 +19,10 @@


if __name__ == "__main__":
    print("======3个票涨停之后买+开盘价>=-3")
    statistic_average(r"C:\Users\Administrator\Desktop\3个票涨停之后买.txt")
    print("======3个票涨停之后买+不限开盘涨幅+3个涨停之后大单打折")
    statistic_average(r"C:\Users\Administrator\Desktop\3个票涨停之后买_不限开盘涨幅.txt")
    # print("======3个票涨停之后买+开盘价>=-3")
    # statistic_average(r"C:\Users\Administrator\Desktop\3个票涨停之后买.txt")
    # print("======3个票涨停之后买+不限开盘涨幅+3个涨停之后大单打折")
    # statistic_average(r"C:\Users\Administrator\Desktop\3个票涨停之后买_不限开盘涨幅.txt")
    strategy_manager.low_suction_strtegy



 strategy/time_series_backtest.py

@@ -586,8 +586,8 @@
            if ticks:
                for tick in ticks:
                    code = tick["symbol"][-6:]
                    # if code not in self.fcodes:
                    #     continue
                    if code not in self.fcodes:
                        continue
                    if DEBUG_CODES and code not in DEBUG_CODES:
                        continue

@@ -648,12 +648,14 @@

                    if not stock_variables.今日最低价 or tick["price"] < stock_variables.今日最低价:
                        stock_variables.今日最低价 = tick["price"]

                    stock_variables.开盘啦最正板块涨停 = most_real_kpl_plate_limit_up_codes_info

                    # compute_result = self.__run_backtest(code, stock_variables)
                    # self.__process_test_result(code, stock_variables, next_trade_day, stock_variables.当前价,
                    #                            time_str, compute_result)
                    if most_real_kpl_plate_limit_up_codes_info:
                        stock_variables.开盘啦最正板块涨停 = most_real_kpl_plate_limit_up_codes_info
                    # if time_str >= '09:30:00':
                    #     if stock_variables.今日大单数据 and stock_variables.开盘啦最正板块涨停 and max(
                    #             [len(stock_variables.开盘啦最正板块涨停.get(x, [])) for x in stock_variables.代码板块]) >= 3:
                    #         compute_result = self.__run_backtest(code, stock_variables)
                    #         self.__process_test_result(code, stock_variables, next_trade_day, stock_variables.当前价,
                    #                                    time_str, compute_result)

                    # if len(real_codes) >= 2 and time_str > '09:30:00':
                    #     # print(time_str, plate)
@@ -766,7 +768,7 @@
            stock_variables.板块成交代码 = self.deal_block_codes


# DEBUG_CODES = ['002365', '000953', '002907', '002688', '003020', '002900', '002082', '000566', '300204', '002317']
# DEBUG_CODES =  ['603040', '603758', '603286', '603586', '605255', '002048', '605208', '002806', '603266', '603788']
DEBUG_CODES = []

VOLUME_LOG_ENABLE = False
@@ -775,7 +777,7 @@

DEBUG_BLOCKS = []

BIG_ORDER_MONEY_THRESHOLD = 200e4
BIG_ORDER_MONEY_THRESHOLD = 100e4

if __name__ == "__main__":
    back_test_dict = {}
@@ -783,7 +785,11 @@
    #         "2025-05-15", "2025-05-16", "2025-05-19", "2025-05-20",  "2025-05-21", "2025-05-22"]
    days = ["2025-05-12", "2025-05-13", "2025-05-14", "2025-05-15", "2025-05-16", "2025-05-19", "2025-05-20",
            "2025-05-21", "2025-05-22", "2025-05-23", "2025-05-26", "2025-05-27", "2025-05-28", "2025-05-29",
            "2025-05-30", "2025-06-03", "2025-06-04", "2025-06-05", "2025-06-06", "2025-06-09", "2025-06-10"]
            "2025-05-30", "2025-06-03", "2025-06-04", "2025-06-05", "2025-06-06", "2025-06-09", "2025-06-10",
            "2025-06-11",  "2025-06-12"]

    # days = ["2025-06-09"]

    days.reverse()
    for day in days:
        if day not in back_test_dict:

 strategy/µÍÎü½Å±¾_±æʶ¶È_v6.py

@@ -85,7 +85,7 @@
    if rate >= settings.avaiable_rates[1] or rate < settings.avaiable_rates[0]:
        return False, f"涨幅不在区间内（{settings.avaiable_rates[0]}-{settings.avaiable_rates[1]}）:{rate}"

    if sv.自由流通市值 > settings.zyltgb_range[1] or sv.自由流通市值 < settings.zyltgb_range[0]:
    if sv.自由流通市值 > settings.zyltgb_range[1] * 1e8 or sv.自由流通市值 < settings.zyltgb_range[0] * 1e8:
        return False, f"自由市值（{sv.自由流通市值}）不满足要求"

    if sv.六个交易日涨幅过高:
@@ -111,11 +111,12 @@

    # ç›®æ ‡ç¥¨æ¿å—涨停个数>=2

    if sv.板块成交代码:
        can_buy_plates -= set(sv.板块成交代码.keys())
    # æ¿å—只能买入一个代码
    # if sv.板块成交代码:
    #     can_buy_plates -= set(sv.板块成交代码.keys())

    if not can_buy_plates:
        return False, "没有涨停的板块"
        return False, f"没有涨停的板块: {[(plate, sv.开盘啦最正板块涨停.get(plate)) for plate in sv.代码板块 if sv.开盘啦最正板块涨停]}  è¿žç»­è€é¢˜æï¼š{sv.连续老题材}"
    # new_plates = set(can_buy_plates) - sv.日出现的板块_2
    # if new_plates:
    #     # æœ‰æ–°é¢˜æï¼Œåˆ¤æ–­æ˜¯å¦è¿‡æ˜¨æ—¥å‰é«˜
@@ -137,7 +138,7 @@
    if sv.今日大单数据:
        # print(sv.今日大单数据[-1][3], format_time(sv.今日大单数据[-1][3]))
        # filter_orders = [(o[0], o[2]) for o in sv.今日大单数据 if format_time(o[3]) >= sv.今日量够信息[0]]
        filter_orders = [(o[0], o[2]) for o in sv.今日大单数据]
        filter_orders = [(o[0], o[2]) for o in sv.今日大单数据 if o[2] >= 200e4]
        filter_orders.reverse()
        orderids = set()
        for o in filter_orders:
@@ -145,19 +146,34 @@
                continue
            orderids.add(o[0])
            big_order_money += o[1]
    big_sell_order_money = 0
    if sv.今日卖大单数据:
        filter_orders = [(o[0], o[2]) for o in sv.今日卖大单数据 if o[2] >= 200e4]
        filter_orders.reverse()
        orderids = set()
        for o in filter_orders:
            if o[0] in orderids:
                continue
            orderids.add(o[0])
            big_sell_order_money += o[1]

    threshold_money = max(sv.自由流通市值 // 1000, 200e4)

    limit_up_codes_count = max([(p, len(sv.开盘啦最正板块涨停.get(p, []))) for p in can_buy_plates], key=lambda x: x[1])[1]

    threshold_money *= max(min(10 - limit_up_codes_count + 1, 10), 5) / 10
    # threshold_money *= max(min(10 - limit_up_codes_count + 1, 10), 5) / 10
    threshold_money *= 0.5

    # print(target_code, sv.自由流通市值, threshold_money, limit_up_codes_count)

    # threshold_money = 200e4  # int(sv.昨日成交量 * 0.2 * sv.今日涨停价 * 0.05)
    if big_order_money < threshold_money:
        return False, f"({big_order_money}/{threshold_money})大单金额不足"

    return True, f" \n\t大单信息：{big_order_money}/{threshold_money}\n\t量够信息：{sv.今日量够信息}\n\t今日最高价:{sv.今日最高价信息} \n\t5日最高价：{sv.日最高价_5}", f"\n\t板块信息：{[(p, sv.开盘啦最正板块涨停.get(p)) for p in can_buy_plates]}", can_buy_plates
    final_big_order_money = big_order_money - big_sell_order_money

    if final_big_order_money < threshold_money:
        return False, f"({round(final_big_order_money / 1e4, 2)}万/{round(threshold_money / 1e4, 2)}万)大单金额不足"

    return True, f" \n\t大单信息：{round(final_big_order_money / 1e4, 2)}万（买：{round(big_order_money / 1e4, 2)}万 å–：{round(big_sell_order_money / 1e4, 2)}万）/{round(threshold_money / 1e4, 2)}万  \n\t量够信息：{sv.今日量够信息}\n\t今日最高价:{sv.今日最高价信息} \n\t5日最高价：{sv.日最高价_5}", f"\n\t板块信息：{[(p, sv.开盘啦最正板块涨停.get(p)) for p in can_buy_plates]}", can_buy_plates


compute_result = can_buy()

 third_data/history_k_data_manager.py

@@ -54,7 +54,7 @@
    for code in codes:
        pre_close = HistoryKDataManager().get_pre_close(code, day)
        if pre_close is not None:
            gpcode_manager.CodePrePriceManager.set_price_pre(code, pre_close, force)
            gpcode_manager.CodePrePriceManager().set_price_pre(code, pre_close, force)
        else:
            not_codes.append(code)
    if not_codes:

 trade/huaxin/huaxin_trade_data_update.py

@@ -65,13 +65,12 @@
                                logger_debug.exception(e)
                            huaxin_trade_record_manager.DealRecordManager.add(datas)
                            if datas:
                                tempList = [
                                    {"time": d["tradeTime"], "type": int(d['direction']), "code": d['securityID']}
                                    for d in datas]
                                try:
                                    trade_manager.process_trade_success_data(tempList)
                                except Exception as e:
                                    logging.exception(e)
                                for d in datas:
                                    if str(d['direction']) != str(huaxin_util.TORA_TSTP_D_Buy):
                                        continue
                                    trade_manager.DealCodesManager().add_deal_order(d['securityID'], d['volume'],
                                                                                    d['price'], d["tradeID"],
                                                                                    d["orderSysID"])
                    # æŒä»“è‚¡
                    elif type_ == "position_list":
                        dataJSON = huaxin_trade_api.get_position_list()

 trade/trade_manager.py

@@ -4,12 +4,14 @@
"""
# äº¤æ˜“管理器
import copy
import json

from code_attribute import gpcode_manager
from db import redis_manager_delegate as redis_manager
from db.mysql_data_delegate import Mysqldb
from db.redis_manager_delegate import RedisUtils
from log_module import async_log_util
from trade import trade_constant


from log_module.log import *
from utils import import_util, tool, huaxin_util
@@ -163,6 +165,83 @@
    # è®¾ç½®äº¤æ˜“账户的可用金额


@tool.singleton
class DealCodesManager:
    """
    æˆäº¤ä»£ç ç®¡ç†
    """
    __db = 12

    # æˆäº¤ä»£ç çš„订单信息：{代码:{交易id:(量，价格，系统订单号)}}

    def __init__(self):
        self.musql = Mysqldb()
        self.__deal_code_orders_info = {}
        self.redis_manager = redis_manager.RedisManager(12)
        # ä¸‹è¿‡å•çš„板块代码
        self.__place_order_plate_codes_info = {}
        self.__load_data()

    def __get_redis(self):
        return self.redis_manager.getRedis()

    def __load_data(self):
        # ä¸ç®—打板的数据
        sql = f"select tradeID,securityID, orderSysID,price,volume from hx_trade_deal_record where `direction` = '0' and tradingDay = '{tool.get_now_date_str('%Y%m%d')}'"
        results = self.musql.select_all(sql)
        if results:
            for r in results:
                self.add_deal_order(r[1], r[4], round(float(r[3]), 2), r[0], r[2])

        val = RedisUtils.get(self.__get_redis(), "place_order_plate_codes_info")
        if val:
            self.__place_order_plate_codes_info = json.loads(val)

    def add_deal_order(self, code, volume, price, trade_id, order_sys_id):
        """
        æ·»åŠ æˆäº¤å¤§å•
        @param code:
        @param volume:
        @param price:
        @param trade_id:
        @param order_sys_id:
        @return:
        """
        pre_price = gpcode_manager.CodePrePriceManager().get_price_pre_cache(code)
        if pre_price and round((price - pre_price) / pre_price, 4) > 0.08 * (tool.get_limit_up_rate(code) - 1) * 10:
            # è§†ä¸ºæ‰“板买入，不处理数据
            return
        if code not in self.__deal_code_orders_info:
            self.__deal_code_orders_info[code] = {}
        if trade_id in self.__deal_code_orders_info[code]:
            return
        self.__deal_code_orders_info[code][trade_id] = (volume, price, order_sys_id)

    def get_deal_codes(self):
        if not self.__deal_code_orders_info:
            return set()
        return set(self.__deal_code_orders_info.keys())

    def place_order(self, plate, code):
        """
        ä¸‹å•
        @param plate:
        @param code:
        @return:
        """
        if plate not in self.__place_order_plate_codes_info:
            self.__place_order_plate_codes_info[plate] = []
        if code not in self.__place_order_plate_codes_info[plate]:
            self.__place_order_plate_codes_info[plate].append(code)
        RedisUtils.setex_async(self.__db, "place_order_plate_codes_info", tool.get_expire(),
                               json.dumps(self.__place_order_plate_codes_info))

    def get_place_order_plate_codes(self):
        return self.__place_order_plate_codes_info


__CodesTradeStateManager = CodesTradeStateManager()


if __name__ == "__main__":
    codes = DealCodesManager().get_codes()
    print(codes)

 utils/init_data_util.py

@@ -19,7 +19,7 @@
        symbol = item['symbol']
        symbol = symbol.split(".")[1]
        pre_close = tool.to_price(decimal.Decimal(str(item['pre_close'])))
        gpcode_manager.CodePrePriceManager.set_price_pre(symbol, pre_close, force)
        gpcode_manager.CodePrePriceManager().set_price_pre(symbol, pre_close, force)


# èŽ·å–è¿‘90天的最大量与最近的量