买入策略修改

Administrator

2023-01-06 59fba698b03a51a8da5b56a919ebbf94d4784f74

 juejin.py

@@ -1,21 +1,44 @@
"""
掘金
"""

from __future__ import print_function, absolute_import

import datetime
import json
import logging
import time as t
import schedule

import gm.api as gmapi

import big_money_num_manager
import client_manager
import code_volumn_manager
import constant
import global_data_loader
import global_util
import gpcode_manager
import threading

import l2_trade_util
import server
import tool

import redis_manager
import authority
import decimal
from l2_code_operate import L2CodeOperate
from log import logger_juejin_tick

redisManager = redis_manager.RedisManager()
import trade_gui
from l2_code_operate import L2CodeOperate
import l2_data_manager_new
from log import logger_juejin_tick, logger_system
from trade_data_manager import CodeActualPriceProcessor
from trade_queue_manager import JueJinBuy1VolumnManager

redisManager = redis_manager.RedisManager(0)
__jueJinBuy1VolumnManager = JueJinBuy1VolumnManager()
__actualPriceProcessor = CodeActualPriceProcessor()


# 设置账户信息
@@ -34,19 +57,87 @@
    return account_id, strategy_id, token


def init_data():
    # 删除之前的分钟级大单撤单数据
    l2_data_manager_new.SecondAverageBigNumComputer.clear_data()
    l2_data_manager_new.AverageBigNumComputer.clear_data()
    # 删除所有的涨停卖数据
    l2_data_manager_new.L2LimitUpSellStatisticUtil.clear()
    # 重置所有的大单数据
    big_money_num_manager.reset_all()
    # 清除水下捞数据
    __actualPriceProcessor.clear_under_water_data()
    # 载入行业股票代码
    global_data_loader.load_industry()
    # 载入代码自由流通市值
    global_data_loader.load_zyltgb()
    # 载入量
    global_data_loader.load_volumn()


# 每日初始化
def everyday_init():
    # 交易時間不能做初始化
    if not tool.is_init_time():
        raise Exception("交易时间不能初始化")

    codes = gpcode_manager.get_gp_list()
    logger_system.info("每日初始化")

    # 今日实时涨停
    global_data_loader.add_limit_up_codes([], True)
    # 主要获取收盘价
    get_latest_info(None)
    # 获取60天最大量与昨日量
    global_util.today_volumn.clear()
    global_util.max60_volumn.clear()
    global_util.yesterday_volumn.clear()
    volumn_dict = get_volumns(codes)
    for key in volumn_dict:
        code_volumn_manager.set_histry_volumn(key, volumn_dict[key][0], volumn_dict[key][1])
    # 清除大单数据
    global_util.big_money_num.clear()
    # 初始化大单数据
    for code in codes:
        big_money_num_manager.add_num(code, 0)
        big_money_num_manager.expire(code)
    # 清除涨停时间
    global_util.limit_up_time.clear()
    init_data()
    # 初始化同花顺主站
    l2_clients = client_manager.getValidL2Clients()
    for client in l2_clients:
        try:
            server.repair_ths_main_site(client)
        except Exception as e:
            pass


def __run_schedule():
    while True:
        schedule.run_pending()


def init(context):
    # gmapi.subscribe(symbols="SZSE.002529", frequency="1d", count=30)

    # 订阅浦发银行, bar频率为一天和一分钟
    # 订阅订阅多个频率的数据，可多次调用subscribe
    # 获取需要监听的股票
    print("掘金初始化")
    init_data()
    logger_system.info("掘金初始化")
    schedule.every().day.at("09:15:00").do(everyday_init)
    t1 = threading.Thread(target=lambda: __run_schedule())
    # 后台运行
    t1.setDaemon(True)
    t1.start()

    # 多个时间点获取收盘价
    gmapi.schedule(schedule_func=get_latest_info, date_rule='1d', time_rule='08:30:00')
    gmapi.schedule(schedule_func=get_latest_info, date_rule='1d', time_rule='08:50:00')
    gmapi.schedule(schedule_func=get_latest_info, date_rule='1d', time_rule='09:15:00')
    gmapi.schedule(schedule_func=get_latest_info, date_rule='1d', time_rule='09:28:00')
    gmapi.schedule(schedule_func=get_current_info, date_rule='1d', time_rule='09:25:00')
    gmapi.schedule(schedule_func=get_current_info, date_rule='1d', time_rule='09:29:00')
    gmapi.schedule(schedule_func=get_current_info, date_rule='1d', time_rule='09:29:35')
    re_subscribe_tick()
    # re_subscribe_bar()

@@ -63,15 +154,16 @@
    for c in clients:
        for i in range(0, 8):
            gpcode_manager.init_listen_code_by_pos(int(c), i)
    data = gpcode_manager.get_gp_list();
    result = JueJinManager.get_gp_latest_info(data);
    codes = gpcode_manager.get_gp_list();
    result = JueJinManager.get_gp_latest_info(codes);
    for item in result:
        sec_level = item['sec_level']
        symbol = item['symbol']
        symbol = symbol.split(".")[1]
        pre_close = tool.to_price(decimal.Decimal(str(item['pre_close'])))
        if sec_level == 1:
            gpcode_manager.set_price_pre(symbol, pre_close)
            if symbol in codes:
                gpcode_manager.set_price_pre(symbol, pre_close)
        else:
            gpcode_manager.rm_gp(symbol)

@@ -80,7 +172,7 @@
def get_current_info():
    data = gpcode_manager.get_gp_list();
    results = JueJinManager.get_gp_current_info(data);
    logger_juejin_tick.debug("定时获取：{}",results)
    logger_juejin_tick.debug("定时获取：{}", results)
    for result in results:
        price = result["price"]
        symbol = result['symbol']
@@ -90,12 +182,12 @@


# 设置收盘价
def set_price_pre(code):
def re_set_price_pre(code):
    codes = [code]
    set_price_pres(codes)
    re_set_price_pres(codes)


def set_price_pres(codes):
def re_set_price_pres(codes):
    result = JueJinManager.get_gp_latest_info(codes);
    for item in result:
        symbol = item['symbol']
@@ -111,13 +203,12 @@
    # print(tick["created_at"])
    relative_timestamp = t.time() % (24 * 60 * 60) + 8 * 60 * 60
    # 9点20-15:05接受数据
    start1 = 60 * 60 * 9 + 24 * 60;
    end1 = 60 * 60 * 11 + 35 * 60;
    start2 = 60 * 60 * 12 + 50 * 60;
    end2 = 60 * 60 * 15 + 5 * 60;
    start1 = 60 * 60 * 9 + 31 * 60
    end1 = 60 * 60 * 11 + 35 * 60
    start2 = 60 * 60 * 12 + 50 * 60
    end2 = 60 * 60 * 15 + 5 * 60
    # TODO 测试
    test = False
    if (start1 < relative_timestamp < end1 or start2 < relative_timestamp < end2) or test:
    if (start1 < relative_timestamp < end1 or start2 < relative_timestamp < end2) or constant.TEST:
        symbol = tick['symbol']
        price = tick['price']
        # print(symbol,price)
@@ -127,28 +218,135 @@

        # 保存最新价
        symbol = symbol.split(".")[1]
        logger_juejin_tick.info("{}   {}    {}".format(symbol, price, tick["created_at"]))
        JueJinManager.add_listen_code(symbol)
        time_ = tick["created_at"].strftime("%H:%M:%S")
        data_ = (symbol, time_, tick["quotes"][0]["bid_v"], tick["quotes"][0]["bid_p"])
        logger_juejin_tick.info("买1量 {},{},{},{}", data_[1], data_[0], data_[2],
                                data_[3])
        # 暂时不采用
        # need_sync = __jueJinBuy1VolumnManager.save(data_[0], data_[1], data_[2], data_[3])
        # if need_sync:
        #     # 同步数据
        #     L2LimitUpMoneyStatisticUtil.verify_num(data_[0], data_[2], data_[1])

        # print(tick["created_at"],tick["quotes"][0]["bid_v"])

        accpt_price(symbol, price)
        __prices_now[symbol] = price


# 获取到现价
def accpt_price(code, price):
def accpt_price(code, price, price_from="juejin"):
    return
    gpcode_manager.set_price(code, price)
    # 获取收盘价
    pricePre = gpcode_manager.get_price_pre(code)
    if pricePre is not None:
        rate = round((price - pricePre) * 100 / pricePre, 1)
        if rate >= 7:
            print(code, price, rate)
            if not gpcode_manager.is_listen(code) and not gpcode_manager.is_operate(
            logger_juejin_tick.info("{}-{}-{}", code, price, rate)
            if not gpcode_manager.is_listen_old(code) and not gpcode_manager.is_operate(
                    code) and not gpcode_manager.is_listen_full():
                L2CodeOperate.get_instance().add_operate(1, code)
                L2CodeOperate.get_instance().add_operate(1, code, "现价变化,rate-{} from-{}".format(rate, price_from))
            # 进入监控
        elif rate < 5:
            # 移除监控
            if gpcode_manager.is_listen(code) and not gpcode_manager.is_operate(code):
                L2CodeOperate.get_instance().add_operate(0, code)
            if gpcode_manager.is_listen_old(code) and not gpcode_manager.is_operate(code):
                L2CodeOperate.get_instance().add_operate(0, code, "现价变化,rate-{} from-{}".format(rate, price_from))


# 获取到现价
def accpt_prices(prices):
    print("价格代码数量：", len(prices))
    __actualPriceProcessor.save_current_price_codes_count(len(prices))
    # 采集的代码数量不对
    if len(gpcode_manager.get_gp_list()) - len(prices) > 2:
        return
    now_str = datetime.datetime.now().strftime("%H:%M:%S")
    now_strs = now_str.split(":")
    now_second = int(now_strs[0]) * 60 * 60 + int(now_strs[1]) * 60 + int(now_strs[2])
    start = 60 * 60 * 9 + 31 * 60
    if False:
        for d in prices:
            code, price = d["code"], float(d["price"])
            accpt_price(code, price, "ths")
    else:
        _code_list = []
        _delete_list = []
        for d in prices:
            code, price = d["code"], float(d["price"])
            gpcode_manager.set_price(code, price)
            # 获取收盘价
            pricePre = gpcode_manager.get_price_pre(code)
            if pricePre is not None:
                rate = round((price - pricePre) * 100 / pricePre, 2)
                if rate >= 0:
                    # 暂存涨幅为正的代码
                    _code_list.append((rate, code))
                else:
                    # 暂存涨幅为负的代码
                    _delete_list.append((rate, code))
                try:
                    __actualPriceProcessor.process_rate(code, rate, now_str)
                except Exception as e:
                    logging.exception(e)

                try:
                    __actualPriceProcessor.save_current_price(code, price,
                                                              gpcode_manager.get_limit_up_price_by_preprice(
                                                                  pricePre) == tool.to_price(
                                                                  decimal.Decimal(d["price"])))
                except Exception as e:
                    logging.exception(e)

        # 排序
        new_code_list = sorted(_code_list, key=lambda e: e.__getitem__(0), reverse=True)
        # 预填充下单代码
        _buy_win_codes = []
        for d in new_code_list:
            _buy_win_codes.append(d[1])
        for d in _delete_list:
            _buy_win_codes.append(d[1])
        try:
            trade_gui.THSBuyWinManagerNew.fill_codes(_buy_win_codes)
        except Exception as e:
            logging.exception(e)
            pass

        client_ids = client_manager.getValidL2Clients()
        # 最多填充的代码数量
        max_count = len(client_ids) * 8
        if max_count == 0:
            max_count = 8
        # 截取前几个代码填充
        add_list = new_code_list[:max_count]
        # 后面的代码全部删除
        _delete_list.extend(new_code_list[max_count:])

        add_code_list = []
        del_list = []
        for d in add_list:
            add_code_list.append(d[1])

        for d in _delete_list:
            del_list.append(d[1])

        # 后面的代码数量
        # 先删除应该删除的代码
        for code in del_list:
            if gpcode_manager.is_listen_old(code):
                # 判断是否在监听里面
                L2CodeOperate.get_instance().add_operate(0, code, "现价变化")
        # 增加应该增加的代码
        for code in add_code_list:
            if not gpcode_manager.is_listen_old(code):
                if not l2_trade_util.is_in_forbidden_trade_codes(code):
                    L2CodeOperate.get_instance().add_operate(1, code, "现价变化")
            else:
                if l2_trade_util.is_in_forbidden_trade_codes(code):
                    L2CodeOperate.get_instance().add_operate(0, code, "现价变化")

        print(add_code_list, del_list)


def on_bar(context, bars):
@@ -175,9 +373,18 @@
def recieve_msg(pipe):
    while True:
        value = pipe.recv()
        print(value)
        if value == 'resub':
        print("跨进程通信：", value)
        jsonValue = json.loads(value)
        action = jsonValue["type"]
        if action == 'resub':
            re_subscribe_tick()
        elif action == 'accpt_price':
            try:
                datas = jsonValue["data"]
                for data in datas:
                    accpt_price(data["code"], float(data["price"]))
            except Exception as e:
                print(str(e))


class JueJinManager:
@@ -219,6 +426,17 @@
    def stop(self):
        gmapi.stop()

    @classmethod
    def add_listen_code(cls, code):
        redis = redisManager.getRedis()
        redis.setex("juejin_listen_code-{}".format(code), 20, "1")

    @classmethod
    def get_listen_codes_lenth(cls):
        redis = redisManager.getRedis()
        keys = redis.keys("juejin_listen_code-*")
        return len(keys)


def trade(code, volume):
    account_id, s_id, token = getAccountInfo()
@@ -229,6 +447,7 @@
    print(result)


# 获取近90天的最大量与最近的量
def get_volumns(codes):
    end = datetime.datetime.now()
    # 获取近90天的历史数据
@@ -248,23 +467,15 @@
        code = result["symbol"].split(".")[1]
        volumn = int(result["volume"])
        day = "{:%Y-%m-%d}".format(result["eob"])
        if _fresult.get(code) is None:
            _fresult[code] = {"max_volumn": volumn, "latest_volumn": volumn}

        if volumn > _fresult[code]["max_volumn"]:
            _fresult[code]["max_volumn"] = volumn;
        _fresult[code]["latest_volumn"] = volumn;
        if _fresult.get(code) is None:
            _fresult[code] = (volumn, volumn)

        if volumn > _fresult[code][0]:
            _fresult[code] = (volumn, _fresult[code][1])
        _fresult[code] = (_fresult[code][0], volumn)
    return _fresult


if __name__ == '__main__':
    # trade("SZSE.000521", 100)
    # print("")
    # JueJinManager.get_gp_latest_info(["000592","002808"])
    get_current_info()

    # data_process.saveCodeVolumn(get_volumns(["000333","002911"]))
    # _redis_manager = redis_manager.RedisManager()
    # redis = _redis_manager.getRedis()
    # keys = redis.keys("test-inrec")
    # print(keys)
    everyday_init()