华鑫适配

Administrator

2023-07-07 7b4dbbfc406509f87b8e6e1842b7b4cb88f97a2d

华鑫适配

 constant.py

@@ -113,3 +113,6 @@

# 掘金本地API
JUEJIN_LOCAL_API = True

# 华鑫L2的卡位数量
HUAXIN_L2_MAX_CODES_COUNT = 30

 l2/l2_data_util.py

@@ -152,11 +152,12 @@
    if len(add_datas) > 0:
        # 保存最近的数据
        __start_time = round(time.time() * 1000)
        redis.setex("l2-data-latest-{}".format(code), tool.get_expire(), json.dumps(datas))
        l2_data_log.l2_time(code, round(time.time() * 1000) - __start_time, "保存最近l2数据用时")
        # 设置进内存
        local_latest_datas[code] = datas
        set_l2_data_latest_count(code, len(datas))
        if datas:
            redis.setex("l2-data-latest-{}".format(code), tool.get_expire(), json.dumps(datas))
            l2_data_log.l2_time(code, round(time.time() * 1000) - __start_time, "保存最近l2数据用时")
            # 设置进内存
            local_latest_datas[code] = datas
            set_l2_data_latest_count(code, len(datas))
        try:
            log.logger_l2_data.info("{}-{}", code, add_datas)
        except Exception as e:

 logs_/log.py

@@ -164,7 +164,7 @@
                   rotation="00:00", compression="zip", enqueue=True)

    def get_path(self, dir_name, log_name):
        return "{}/logs_/gp/{}/{}".format('D:' if constant.is_windows() else '/home', dir_name,
        return "{}/logs/gp/{}/{}".format('D:' if constant.is_windows() else '/home', dir_name,
                                         log_name) + ".{time:YYYY-MM-DD}.log"

    def get_logger(self, log_name):

 output/code_info_output.py

@@ -540,8 +540,11 @@

    return break_time, records_new, records_new_data


if __name__ == '__main__':
    plate_info = kpl_api.getStockIDPlate("600748")
    print(plate_info)

if __name__ == '__main__1':
    datas = {
        "base_url": "http://192.168.3.122/kp/",
        "code_name": "天域生态 002255",

 third_data/kpl_api.py

@@ -23,9 +23,10 @@
        raise Exception("请求出错")
    result = response.text
    result = json.loads(result)
    print(result)
    if int(result["errcode"]) != 0:
        return None
    return result["ListJX"]
    return result["ListJX"] if result["ListJX"] else result["List"]

# 获取概念代码
def getCodesByPlate(plate_code):

 trade/current_price_process_manager.py

@@ -8,6 +8,7 @@
from ths import client_manager
import constant
from code_attribute import gpcode_manager
from trade.huaxin import huaxin_trade_api
from utils import tool, import_util
from ths.l2_code_operate import L2CodeOperate
from trade import trade_manager, l2_trade_util
@@ -67,26 +68,30 @@
        # -------------------------------处理交易位置分配---------------------------------
        # 排序
        new_code_list = sorted(_code_list, key=lambda e: (e.__getitem__(2), 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:
            if not constant.API_TRADE_ENABLE and trade_gui is not None:
                trade_gui.THSBuyWinManagerNew.fill_codes(_buy_win_codes)
        except Exception as e:
            logging.exception(e)
            pass
        if constant.L2_SOURCE_TYPE == constant.L2_SOURCE_TYPE_THS:
            # 预填充下单代码
            _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:
                if not constant.API_TRADE_ENABLE and trade_gui is not None:
                    trade_gui.THSBuyWinManagerNew.fill_codes(_buy_win_codes)
            except Exception as e:
                logging.exception(e)
                pass

        # -------------------------------处理L2监听---------------------------------

        client_ids = client_manager.getValidL2Clients()
        # 最多填充的代码数量
        max_count = len(client_ids) * constant.L2_CODE_COUNT_PER_DEVICE
        if max_count == 0:
            max_count = constant.L2_CODE_COUNT_PER_DEVICE
        max_count = 0
        if constant.L2_SOURCE_TYPE == constant.L2_SOURCE_TYPE_THS:
            client_ids = client_manager.getValidL2Clients()
            # 最多填充的代码数量
            max_count = len(client_ids) * constant.L2_CODE_COUNT_PER_DEVICE
            if max_count == 0:
                max_count = constant.L2_CODE_COUNT_PER_DEVICE
        else:
            max_count = constant.HUAXIN_L2_MAX_CODES_COUNT

        _delete_list = []
        for item in new_code_list:
@@ -108,32 +113,44 @@
        for d in _delete_list:
            del_code_list.append(d[1])

        # 后面的代码数量
        # 先删除应该删除的代码
        for code in del_code_list:
            if gpcode_manager.is_listen_old(code):
                cid, pid = gpcode_manager.get_listen_code_pos(code)
                # 强制移除
                if cid and pid:
                    gpcode_manager.set_listen_code_by_pos(cid, pid, "")
                # 判断是否在监听里面
                L2CodeOperate.get_instance().add_operate(0, code, "现价变化")
        # 增加应该增加的代码
        for code in add_code_list:
            if not gpcode_manager.is_listen_old(code):
                L2CodeOperate.get_instance().add_operate(1, code, "现价变化")
        if constant.L2_SOURCE_TYPE == constant.L2_SOURCE_TYPE_HUAXIN:
            # 华鑫L2,获取加入代码的涨停价
            add_datas = []
            for d in add_code_list:
                limit_up_price = gpcode_manager.get_limit_up_price(d)
                min_volume = 50 * 10000 // limit_up_price
                add_datas.append((d, min_volume, limit_up_price))
                try:
                    huaxin_trade_api.set_l2_codes_data(add_datas)
                except Exception as e:
                    logging.exception(e)
        else:
            # 后面的代码数量
            # 先删除应该删除的代码
            for code in del_code_list:
                if gpcode_manager.is_listen_old(code):
                    cid, pid = gpcode_manager.get_listen_code_pos(code)
                    # 强制移除
                    if cid and pid:
                        gpcode_manager.set_listen_code_by_pos(cid, pid, "")
                    # 判断是否在监听里面
                    L2CodeOperate.get_instance().add_operate(0, code, "现价变化")
            # 增加应该增加的代码
            for code in add_code_list:
                if not gpcode_manager.is_listen_old(code):
                    L2CodeOperate.get_instance().add_operate(1, code, "现价变化")

        # 获取卡位数量
        free_count = gpcode_manager.get_free_listen_pos_count()
        if free_count < 2:
            # 空闲位置不足
            listen_codes = gpcode_manager.get_listen_codes()
            for code in listen_codes:
                if not gpcode_manager.is_in_gp_pool(code):
                    client_id, pos = gpcode_manager.get_listen_code_pos(code)
                    gpcode_manager.set_listen_code_by_pos(client_id, pos, "")
                    free_count += 1
                    if free_count > 2:
                        break
            # 获取卡位数量
            free_count = gpcode_manager.get_free_listen_pos_count()
            if free_count < 2:
                # 空闲位置不足
                listen_codes = gpcode_manager.get_listen_codes()
                for code in listen_codes:
                    if not gpcode_manager.is_in_gp_pool(code):
                        client_id, pos = gpcode_manager.get_listen_code_pos(code)
                        gpcode_manager.set_listen_code_by_pos(client_id, pos, "")
                        free_count += 1
                        if free_count > 2:
                            break

        print(add_code_list, del_code_list)
            print(add_code_list, del_code_list)

 trade/huaxin/trade_server.py

@@ -7,8 +7,13 @@
import socketserver
import threading
import time
import l2

import constant
from code_attribute import gpcode_manager
from l2 import l2_data_manager_new, l2_data_log
from l2.huaxin import l2_huaxin_util
from logs_.log import logger_l2_error
from trade.huaxin.huaxin_log import logger_l2_orderdetail, logger_l2_transaction, logger_l2_upload, \
    logger_contact_debug, logger_trade_callback, logger_trade_debug

@@ -147,6 +152,19 @@
                        code = data["code"]
                        datas = data["data"]
                        logger_l2_orderdetail.info(f"{code}#{datas}")
                        origin_start_time = round(time.time() * 1000)
                        try:
                            # 转换数据格式
                            datas = l2_huaxin_util.get_format_l2_datas(code, datas, gpcode_manager.get_limit_up_price(code))
                            __start_time = round(time.time() * 1000)
                            l2_data_manager_new.L2TradeDataProcessor().process_add_datas(code, datas, 0, __start_time)
                        except Exception as e:
                            logger_l2_error.exception(e)
                        finally:
                            l2_data_log.l2_time(code, round(time.time() * 1000) - origin_start_time,
                                                "l2数据处理总耗时",
                                                True)
                            l2.l2_data_util.save_l2_data(code, None, datas)
                        sk.sendall(json.dumps({"code": 0}).encode(encoding='utf-8'))

                    elif data_json["type"] == "l2_trans":

 trade/trade_huaxin.py

New file
@@ -0,0 +1,190 @@
"""
华鑫交易
"""
import json
import logging
import time

import gm.api as gmapi

import constant
from code_attribute import gpcode_manager
from db.redis_manager import RedisManager
from logs_.log import logger_juejin_trade
from utils import network_util, tool

__context_dict = {}

account_id = "77916efb-b856-46ee-9680-71be0fe18a42"
token = "38fb624832c1949708c7600abaf1e863d27663b3"
gmapi.set_token(token)


# gmapi.set_account_id(account_id)


# 交易订单号管理
class TradeOrderIdManager:
    __redisManager = RedisManager(2)

    @classmethod
    def __get_redis(cls):
        return cls.__redisManager.getRedis()

    # 添加订单ID
    @classmethod
    def add_order_id(cls, code, account_id, order_id):
        cls.__get_redis().sadd(f"huaxin_order_id-{code}", json.dumps((account_id, order_id)))
        cls.__get_redis().expire(f"huaxin_order_id-{code}", tool.get_expire())

    # 删除订单ID
    @classmethod
    def remove_order_id(cls, code, account_id, order_id):
        cls.__get_redis().srem(f"huaxin_order_id-{code}", json.dumps((account_id, order_id)))

    # 查询所有的订单号
    @classmethod
    def list_order_ids(cls, code):
        return cls.__get_redis().smembers(f"huaxin_order_id-{code}")


def init(context):
    __context_dict["init"] = context
    print("掘金交易初始化成功")


# 可用金额
def get_account_left_money():
    if "init" in __context_dict:
        dict_ = __context_dict["init"].account().cash
        return dict_["available"]
    return None


# 通过量下单,返回(代码，账号ID，订单号)
def order_volume(code, price, count):
    if not constant.TRADE_ENABLE:
        return
    if code.find("00") != 0 and code.find("60") != 0:
        raise Exception("只支持00开头与60开头的代码下单")
    code_str = code
    if code[0:2] == '00':
        code_str = f"SZSE.{code}"
    elif code[0:2] == '60':
        code_str = f"SHSE.{code}"
    start_time = time.time()
    results = gmapi.order_volume(code_str, count, gmapi.OrderSide_Buy, gmapi.OrderType_Limit, gmapi.PositionEffect_Open,
                                 price=price,
                                 order_duration=gmapi.OrderDuration_GFD, account=account_id)
    print("掘金下单耗时", time.time() - start_time)
    logger_juejin_trade.info(f"{code}：下单耗时{round(time.time() - start_time, 3)}s")

    if results:
        print("下单结果", results)
        result = results[0]
        if result["ord_rej_reason_detail"]:
            logger_juejin_trade.info(f"{code}：下单失败：{result['ord_rej_reason_detail']}")
            raise Exception(result["ord_rej_reason_detail"])
        else:
            TradeOrderIdManager.add_order_id(code, result["account_id"], result["cl_ord_id"])
            logger_juejin_trade.info(f"{code}：下单成功 ord_id:{result['cl_ord_id']}")
            return result["symbol"].split(".")[1], result["account_id"], result["cl_ord_id"]
    else:
        raise Exception("下单失败，无返回")


# 撤单
def cancel_order(code):
    orders_info = TradeOrderIdManager.list_order_ids(code)
    orders = []
    if orders_info:
        for order in orders_info:
            order_info = json.loads(order)
            orders.append({'cl_ord_id': order_info[1], 'account_id': order_info[0]})
    if orders:
        logger_juejin_trade.info(f"{code}：开始执行撤单")
        # 执行3次撤单
        for i in range(3):
            gmapi.order_cancel(orders)
        logger_juejin_trade.info(f"{code}：撤单成功，撤单数量：{len(orders)}")
        for order in orders:
            TradeOrderIdManager.remove_order_id(code, order["account_id"], order["cl_ord_id"])


# 撤单
def __cancel_order(account_id, cl_ord_id):
    orders = [{'cl_ord_id': cl_ord_id, 'account_id': account_id}]
    gmapi.order_cancel(orders)


def test():
    symbols = gpcode_manager.get_gp_list_with_prefix(["002531"])
    data = gmapi.get_instruments(symbols=",".join(symbols))
    print(data)


def run():
    print("启动读取掘金交易数据")
    # strategy_id = "e97a257e-1bba-11ed-a1b1-00e070c694ff"
    # token = "a2eed2b159e9238dc0353fc3e73734d7677f7baf"
    # gmapi.run(strategy_id, filename="trade.trade_juejin.py", mode=gmapi.MODE_LIVE, token=token)
    while True:
        try:
            if tool.is_trade_time():
                datas = get_execution_reports()
                # 上传数据
                fdatas = []
                for d in datas:
                    fdatas.append(
                        {"code": d[0], "money": d[4], "num": d[2], "price": d[3], "time": d[7], "trade_num": d[5],
                         "type": d[1] - 1})
                if fdatas:
                    network_util.send_socket_msg("127.0.0.1", 9001, {"type": 3, "data": fdatas})
        except Exception as e:
            logging.exception(e)
        # 2s更新
        time.sleep(1.5)


# 获取成交列表,返回的内容为:[(代码,买(1)/卖(2),量,价格,成交金额,订单ID,委托客户端ID,成交时间,成交日期)]
def get_execution_reports():
    gmapi.set_account_id(account_id)
    reports = gmapi.get_execution_reports()
    results = []
    for r in reports:
        if not r['ord_rej_reason_detail']:
            results.append(
                [r["symbol"].split(".")[1], r["side"], r["volume"], round(r["price"], 2), round(r["amount"], 2),
                 r["order_id"],
                 r["cl_ord_id"], r["created_at"].strftime("%H:%M:%S"), r["created_at"].strftime("%Y-%m-%d")])
    # 根据订单号合并数据
    temp_dict = {}
    for r in results:
        if r[5] not in temp_dict:
            temp_dict[r[5]] = r
        else:
            temp_dict[r[5]][2] += r[2]
            temp_dict[r[5]][4] += r[4]
    results = [temp_dict[k] for k in temp_dict]
    print("获取已成交数量：", len(results))

    return results


if __name__ == "__main__":
    datas = get_execution_reports()
    # 上传数据
    fdatas = []
    for d in datas:
        fdatas.append(
            {"code": d[0], "money": d[4], "num": d[2], "price": d[3], "time": d[7], "trade_num": d[5],
             "type": d[1] - 1})
    print(fdatas)
    network_util.send_socket_msg("127.0.0.1", 9001, {"type": 3, "data": fdatas})
    # print(order_volume("000566", 4.66, 100))
    # gmapi.set_token(token)
    # gmapi.set_account_id(account_id)
    # cancel_order("000566")
    # orders=[]
    # orders.append({'cl_ord_id':"3a691f3d-fdc7-11ed-838e-f4b5203f67bf", 'account_id': "8099a935-a991-4871-977f-206c6d3e04ca"})
    # gmapi.order_cancel(orders)

 trade/trade_juejin.py

@@ -1,5 +1,5 @@
"""
同花顺交易操作工具
 掘金交易
"""
import json
import logging