L1数据处理流程优化

admin

2025-04-11 1e29252aa811815f202a4d31f264b19c8d76def3

L1数据处理流程优化

 main.py

@@ -20,7 +20,7 @@
    order_methods, local_data_management, kpl_data_manager, market_sentiment_analysis, plate_strength_analysis, \
    selling_strategy
from huaxin_client import l2_market_client, l2_client
from log_module import async_log_util
from log_module import async_log_util, log
from trade import huaxin_trade_data_update, huaxin_trade_api
from utils import hx_qc_value_util, huaxin_util, juejin_api, tool

@@ -225,6 +225,7 @@

# 第三步：执行策略的初始设置
if __name__ == '__main__':
    log.close_print()
    class MyMarketDataCallback(l2_market_client.L2MarketDataCallback):
        def on_markets(self, datas):
            """
@@ -233,8 +234,6 @@
            :return:
            """
            data_cache.latest_code_market_info_dict = {x[0]: x for x in datas}
            if datas:
                print(datas[0])
            if constant.is_windows():
                instant_time_market.get_current_info()
            else:

 strategy/buying_strategy.py

@@ -203,8 +203,9 @@
                # 查询当前个股的代码是否在板块强度个股列表中（低吸优选目标）
                strength_list_have_it = False  # 为查询结果赋初值
                strength_plate = []
                market_sift_plate_stock_dict = data_cache.market_sift_plate_stock_dict
                # 遍历字典的每个键值对
                for key, values in data_cache.market_sift_plate_stock_dict.items():
                for key, values in market_sift_plate_stock_dict.items():
                    # 遍历每个键对应的列表中的列表
                    for stock_list in values:
                        if len(stock_list) > 1 and symbol_code in stock_list[1]:

 strategy/instant_time_market.py

@@ -58,8 +58,6 @@
# 调用所有以current信息为核心策略的函数
def strategic_thread_manager(current_info):
    if current_info is not None:
        symbol_code = current_info[0]
        data_cache.current_l1_dict[symbol_code] = current_info
        # 调用交易策略模块中的涨幅视界策略
        # 买入策略调用
        buying_strategy.growth_view_strategy(current_info)
@@ -296,6 +294,11 @@
            #         print(f"i===={i}")
            get_all_stocks_current_open(current_infos)
            get_all_stocks_current_high_and_low(current_infos)

            # 保存现价
            for current_info in current_infos:
                data_cache.current_l1_dict[current_info[0]] = current_info

            for current_info in current_infos:
                try:
                    if current_info is not None:
@@ -343,6 +346,10 @@
        get_all_stocks_current_open(current_infos)
        get_all_stocks_current_high_and_low(current_infos)
        if current_infos:
            # 保存现价
            for current_info in current_infos:
                data_cache.current_l1_dict[current_info[0]] = current_info

            # 分批处理数据
            ds = []
            total_count = len(current_infos)

 strategy/market_sentiment_analysis.py

@@ -465,14 +465,10 @@

        """计算市场关键指标因子"""
        factors = {}
        # 总票数
        total_stocks = zero + rise_one + rise_two + rise_three + rise_four + rise_five + rise_six + rise_seven + rise_eight + rise_nine + rise_ten + fall_one + fall_two + fall_three + fall_four + fall_five + fall_six + fall_seven + fall_eight + fall_nine + fall_ten
        print(f"total_stocks = {total_stocks}")
        factors['total_stocks'] = total_stocks
        # 涨跌统计（排除0%）
        rise = rise_one + rise_two + rise_three + rise_four + rise_five + rise_six + rise_seven + rise_eight + rise_nine + rise_ten
        fall = fall_one + fall_two + fall_three + fall_four + fall_five + fall_six + fall_seven + fall_eight + fall_nine + fall_ten

        # 涨跌统计
        calm_dirt = {
            'zero': zero
        }
        rise_dirt = {
            'rise_one': rise_one,
            'rise_two': rise_two,
@@ -497,21 +493,38 @@
            'fall_nine': fall_nine,
            'fall_ten': fall_ten,
        }
        # 计算所有值的总和
        total_sum = sum(rise_dirt.values())
        # 计算每个键的值占总和的百分比
        percentages = {key: round((value / total_sum) * 100, 2) for key, value in rise_dirt.items()}
        rise_and_fall_dirt = {}
        rise_and_fall_dirt.update(calm_dirt)
        rise_and_fall_dirt.update(rise_dirt)
        rise_and_fall_dirt.update(fall_dirt)
        # 计算涨跌统计所有股票数量总和
        rise_and_fall_sum = sum(rise_and_fall_dirt.values())
        # 总票数
        factors['total_stocks'] = rise_and_fall_sum
        # 计算上涨股票数量的和
        rise_sum = sum(rise_dirt.values())
        # 计算上涨股票数量的和
        fall_sum = sum(fall_dirt.values())
        # 计算每个键（每个幅段）的值占总和的百分比
        percentages = {key: round((value / rise_and_fall_sum) * 100, 2) for key, value in rise_and_fall_dirt.items()}
        # # 计算每个键（每个涨幅段）的值占总和的百分比
        # rise_percentages = {key: round((value / total_sum) * 100, 2) for key, value in rise_dirt.items()}
        # # 计算每个键（每个涨幅段）的值占总和的百分比
        # fall_percentages = {key: round((value / total_sum) * 100, 2) for key, value in fall_dirt.items()}
        # 找到最大值对应的键
        max_key = max(rise_dirt, key=rise_dirt.get)
        max_key = max(rise_and_fall_dirt, key=rise_and_fall_dirt.get)
        # 找到最小值对应的键
        min_key = min(rise_dirt, key=rise_dirt.get)
        min_key = min(rise_and_fall_dirt, key=rise_and_fall_dirt.get)

        # 涨跌比因子 ---------------------------------------------------
        factors['rise_vs_fall'] = {
            'rise_vs_fall_ratio': rise / fall if fall > 0 else 0,  # 涨跌比
            'rise_vs_fall_ratio': round(rise_sum / fall_sum, 2) if fall_sum > 0 else 0,  # 涨跌比
            'rise_gather_area': max_key,  # 找到最大值
            'rise_scattered_area': min_key,  # 找到最小值
            'percentages': percentages,  # 找到最小值
            'percentages': percentages,  # 各个涨幅段的股票分布比例
            # 'rise_percentages': rise_percentages,  # 各个涨幅段的股票分布比例
            # 'fall_percentages': fall_percentages,  # 各个跌幅段的股票分布比例
            'rise_and_fall_sum': rise_and_fall_sum
        }

        # 资金流向因子 --------------------------------------------------
@@ -523,16 +536,16 @@

        # 市场情绪因子 --------------------------------------------------
        factors['sentiment'] = {
            'zt_ratio': data['ZT'] / total_stocks,  # 涨停占总比
            'zt_ratio': data['ZT'] / rise_and_fall_sum,  # 涨停占总比
            'zt_vs_dt_ratio': data['ZT'] / data['DT'],  # 涨停跌停比
            'sign': 1 if '较好' in data['sign'] else 0,  # 情绪标签量化
            'rise_vs_all_stocks_ratio': data['SZJS'] / total_stocks,  # 活跃股占比
            'rise_vs_all_stocks_ratio': data['SZJS'] / rise_and_fall_sum,  # 活跃股占比
            'rise_vs_fall_ratio': data['SZJS'] / data['SZJS'],  # 活跃股占比
            'extreme_ratio': (data['ZT'] + data['DT']) / total_stocks,  # 极端波动（涨停+跌停占总比）
            'extreme_ratio': (data['ZT'] + data['DT']) / rise_and_fall_sum,  # 极端波动（涨停+跌停占总比）
        }
        return factors

#

# # ====================== 策略信号生成 ======================
def generate_signals(factors):
    """生成多维度交易信号"""
@@ -671,8 +684,9 @@
    print(f"资金净流入(元): {factors['capital_flow']['net']:,}")
    print(f"涨停股占比: {factors['sentiment']['zt_ratio']:.2%}")
    print(f"市场情绪量化: {'积极' if factors['sentiment']['sign'] else '谨慎'}")
    print(f"上涨聚集区域：{factors['rise_vs_fall']['rise_gather_area']}")
    print(f"上涨零散区域：{factors['rise_vs_fall']['percentages']}")
    print(f"聚集区域：{factors['rise_vs_fall']['rise_gather_area']}")
    print(f"零散区域：{factors['rise_vs_fall']['rise_scattered_area']}")
    print(f"涨跌因子字典={factors['rise_vs_fall']}")

    print("\n========== 策略信号 ==========")
    for i, signal in enumerate(signals, 1):

 strategy/selling_strategy.py

@@ -138,12 +138,16 @@
    # :return: {'code': 0, 'data': [{'investorID': '00044396', 'tradingDay': '20250109', 'securityName': '海得控制', 'securityID': '002184', 'historyPos': 0, 'historyPosFrozen': 0, 'todayBSPos': 1000, 'todayBSPosFrozen': 0, 'historyPosPrice': 0.0, 'totalPosCost': 20316.176271, 'prePosition': 0, 'availablePosition': 0, 'currentPosition': 1000, 'openPosCost': 20310.0, 'todayCommission': 6.176271, 'todayTotalBuyAmount': 20310.0, 'todayTotalSellAmount': 0.0}, {'investorID': '00044396', 'tradingDay': '20250109', 'securityName': '省广集团', 'securityID': '002400', 'historyPos': 0, 'historyPosFrozen': 0, 'todayBSPos': 0, 'todayBSPosFrozen': 0, 'historyPosPrice': 0.0, 'totalPosCost': 0.0, 'prePosition': 0, 'availablePosition': 0, 'currentPosition': 0, 'openPosCost': 0.0, 'todayCommission': 0.0, 'todayTotalBuyAmount': 8230.0, 'todayTotalSellAmount': 0.0}]}

    # print(f"data_cache.account_positions_dict==={data_cache.account_positions_dict}")
    if data_cache.account_positions_dict is not None:
        for index, element in enumerate(data_cache.account_positions_dict):
    account_positions_dict = data_cache.account_positions_dict

    if account_positions_dict is not None:
        for index, element in enumerate(account_positions_dict):
            # print(f"index=={index}")
            # print(f"element=={element}")
            element_symbol = basic_methods.format_stock_symbol(element['securityID'])
            if symbol_code == element['securityID']:  # 当循环到的持仓代码正好是current_data的代码时，即取得了持仓个股的行情快照数据
            if symbol_code != element['securityID']:
                continue
            # 当循环到的持仓代码正好是current_data的代码时，即取得了持仓个股的行情快照数据
                # logger_info(f"index = {index}")
                # logger_info(f"element = {element}")
                account_positions_symbol = element_symbol  # 当前循环到的持仓个股代码