Administrator
2024-11-25 7499d1aab63c5a97bc416f28a7e44d9e07b0ec65
更新板块流入流出统计
4个文件已修改
154 ■■■■■ 已修改文件
l2/l2_data_manager_new.py 2 ●●● 补丁 | 查看 | 原始文档 | blame | 历史
l2_test.py 63 ●●●●● 补丁 | 查看 | 原始文档 | blame | 历史
third_data/custom_block_in_money_manager.py 86 ●●●●● 补丁 | 查看 | 原始文档 | blame | 历史
trade/buy_radical/radical_buy_data_manager.py 3 ●●●● 补丁 | 查看 | 原始文档 | blame | 历史
l2/l2_data_manager_new.py
@@ -2181,7 +2181,7 @@
            # 如果板上放量不可买入就需要删除信号
            if not constant.CAN_RADICAL_BUY_AT_LIMIT_UP and code in RadicalBuyDealCodesManager.buy_by_l2_delegate_expire_time_dict:
                RadicalBuyDealCodesManager.buy_by_l2_delegate_expire_time_dict.pop(code)
            return True, result[1], radical_data[2], radical_data[3], result[3]
            return True, result[1], radical_data[2], radical_data[4], result[3]
        return result
    @classmethod
l2_test.py
@@ -1,15 +1,60 @@
import http
import json
import logging
import multiprocessing
import socketserver
import threading
import time
from http.server import BaseHTTPRequestHandler
import psutil
import requests
from huaxin_client import l2_client_test, l1_subscript_codes_manager
from log_module.log import logger_local_huaxin_l2_transaction_big_order
from log_module.log import logger_local_huaxin_l2_transaction_big_order, logger_system
from third_data.custom_block_in_money_manager import CodeInMoneyManager, BlockInMoneyRankManager
from utils import tool
import urllib.parse as urlparse
from urllib.parse import parse_qs
class DataServer(BaseHTTPRequestHandler):
    # 禁用日志输出
    def log_message(self, format, *args):
        pass
    def do_GET(self):
        path = self.path
        url = urlparse.urlparse(path)
        response_data = ""
        if url.path == "/get_block_codes_money":
            ps_dict = dict([(k, v[0]) for k, v in parse_qs(url.query).items()])
            block = ps_dict['block']
            try:
                fdatas = BlockInMoneyRankManager().get_block_codes_money(block)
                response_data = json.dumps({"code": 0, "data": fdatas})
            except Exception as e:
                response_data = json.dumps({"code": 1, "msg": str(e)})
        self.send_response(200)
        # 发给请求客户端的响应数据
        self.send_header('Content-type', 'application/json')
        self.end_headers()
        self.wfile.write(response_data.encode())
class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    pass
def __run_server(addr, port):
    handler = DataServer
    try:
        httpd = ThreadedHTTPServer((addr, port), handler)
        print("L2_IN_MONEY HTTP server is at: http://%s:%d/" % (addr, port))
        httpd.serve_forever()
    except Exception as e:
        logger_system.exception(e)
        logger_system.error(f"端口服务器:{port} 启动失败")
def run():
@@ -67,9 +112,9 @@
            BlockInMoneyRankManager().compute()
            in_list = BlockInMoneyRankManager().get_in_list()
            out_list = BlockInMoneyRankManager().get_out_list()
            # (代码,名称,强度,主力净额)
            fins = [(0, x[0], 0, x[1]) for x in in_list[:50]]
            fouts = [(0, x[0], 0, x[1]) for x in out_list[:50]]
            # (代码,名称,强度,主力净额)
            # 上传
            __upload_data("jingxuan_rank", json.dumps(fins))
            __upload_data("jingxuan_rank_out", json.dumps(fouts))
@@ -80,8 +125,22 @@
            time.sleep(3)
def __update_today_limit_up_records():
    while True:
        try:
            BlockInMoneyRankManager().load_today_limit_up_codes()
        except:
            pass
        finally:
            time.sleep(3)
if __name__ == "__main__":
    threading.Thread(target=__compute_and_upload, daemon=True).start()
    # 启动内部接口服务
    threading.Thread(target=__run_server, args=("0.0.0.0", 9005,), daemon=True).start()
    # 启用定时更新当日涨停
    threading.Thread(target=__update_today_limit_up_records, daemon=True).start()
    run()
    while True:
        time.sleep(2)
third_data/custom_block_in_money_manager.py
@@ -2,9 +2,11 @@
自定义板块流入金额
"""
import copy
import itertools
import os
import constant
from db import mysql_data_delegate as mysql_data
from huaxin_client import l1_subscript_codes_manager
from third_data.kpl_data_constant import LimitUpCodesBlockRecordManager
from third_data.third_blocks_manager import BlockMapManager
@@ -53,11 +55,15 @@
    """
    板块流入流出管理
    """
    __mysql_db = mysql_data.Mysqldb()
    __code_blocks = {}
    __in_list = []
    __out_list = []
    # 最近这段时间的代码涨停次数
    __history_code_limit_up_count = {}
    # 被排除的代码
    __exclude_codes = set()
    def __load_codes(self):
        codes = []
@@ -79,9 +85,59 @@
                    fblocks -= constant.KPL_INVALID_BLOCKS
                self.__code_blocks[code] = fblocks
    def __load_exclude_codes(self):
        """
        获取之前4个交易日的数据
        @return:
        """
        max_day = tool.get_now_date_str()
        min_day = tool.date_sub(max_day, 30)
        sql = f"select * from (select distinct(r.`_day`) as 'day' from `kpl_limit_up_record` r where r.`_day`<'{max_day}' and r.`_day`>'{min_day}') a order by a.day desc limit 4"
        results = self.__mysql_db.select_all(sql)
        dates = [r[0] for r in results]
        # 获取之前4天涨停次数>=2次的代码
        day_codes = {}
        for day in dates:
            sql = f"select distinct(r._code) from kpl_limit_up_record r where r.`_day`='{day}'"
            results = self.__mysql_db.select_all(sql)
            day_codes[day] = set([x[0] for x in results])
        codes_list = [day_codes[k] for k in day_codes]
        # 统计代码的涨停天数
        code_limit_up_count_dict = {}
        for codes in codes_list:
            for c in codes:
                if c not in code_limit_up_count_dict:
                    code_limit_up_count_dict[c] = 0
                code_limit_up_count_dict[c] += 1
        self.__history_code_limit_up_count = code_limit_up_count_dict
        self.load_today_limit_up_codes()
    def load_today_limit_up_codes(self):
        # 加载今日涨停代码
        day = tool.get_now_date_str()
        sql = f"select distinct(r._code) from kpl_limit_up_record r where r.`_day`='{day}'"
        results = self.__mysql_db.select_all(sql)
        codes = set([x[0] for x in results])
        # 计算需要排除的代码
        temp_limit_up_count = {}
        # 统计总共涨停天数
        for c in self.__history_code_limit_up_count:
            if c in codes:
                temp_limit_up_count[c] = self.__history_code_limit_up_count[c] + 1
            else:
                temp_limit_up_count[c] = self.__history_code_limit_up_count[c]
        exclude_codes = set()
        for c in temp_limit_up_count:
            if temp_limit_up_count[c] < 3:
                continue
            exclude_codes.add(c)
        self.__exclude_codes = exclude_codes
    def __init__(self):
        self.codes = self.__load_codes()
        self.__load_blocks()
        self.__load_exclude_codes()
        print("排除的代码", self.__exclude_codes)
    def get_codes(self):
        return self.codes
@@ -90,6 +146,8 @@
        codes = self.codes
        block_money = {}
        for code in codes:
            if code in self.__exclude_codes:
                continue
            money = CodeInMoneyManager().get_money(code)
            if money is None:
                continue
@@ -110,6 +168,29 @@
        temp_list.sort(key=lambda x: x[1])
        self.__out_list = temp_list
    def get_block_codes_money(self, block):
        """
        获取板块中代码的流入流出
        @param block:
        @return:(代码, 金额, 是否被排除)
        """
        codes = self.codes
        fdatas = []
        for code in codes:
            if code in self.__exclude_codes:
                continue
            money = CodeInMoneyManager().get_money(code)
            if money is None:
                continue
            before_fblocks = LimitUpCodesBlockRecordManager().get_radical_buy_blocks(code)
            if not before_fblocks:
                before_fblocks = set()
            fblocks = BlockMapManager().filter_blocks(before_fblocks)
            if block not in fblocks:
                continue
            fdatas.append((code, money, code in self.__exclude_codes))
        return fdatas
    def get_in_list(self):
        return self.__in_list
@@ -118,7 +199,8 @@
if __name__ == '__main__':
    print(CodeInMoneyManager().get_money("300264"))
    BlockInMoneyRankManager()
    # print(CodeInMoneyManager().get_money("300264"))
    # BlockInMoneyRankManager().compute()
    # print(BlockInMoneyRankManager().get_in_list()[:20])
    # print(BlockInMoneyRankManager().get_out_list()[:20])
trade/buy_radical/radical_buy_data_manager.py
@@ -650,7 +650,7 @@
        @return:
        """
        block_codes = BlockSpecialCodesManager().get_block_codes(block)
        if code not in block_codes:
        if not block_codes or code not in block_codes:
            return False, "无辨识度"
        history_index, history_before_codes_info = cls.__get_history_index(code, block, yesterday_limit_up_codes)
        if history_index > 5:
@@ -737,7 +737,6 @@
        @param code:
        @return: {激进买的板块}, 原因
        """
        # 计算
        # 获取开1的板块
        open_limit_up_code_dict = kpl_data_constant.open_limit_up_code_dict_for_radical_buy