增加从data_server获取数据接口

Administrator

2023-10-09 b893d87faa84338e5494bcb06551b958ff2b3510

增加从data_server获取数据接口

 huaxin_client/constant.py

@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*-
# addr, port = "111.230.16.67", 10008
# SERVER_IP = '43.138.167.68'
# SERVER_IP = '192.168.3.122'
SERVER_IP = "43.138.167.68"
SERVER_PORT = 10008
TEST = True

 outside_api_command_manager.py

@@ -53,7 +53,7 @@
API_TYPE_GET_ENV = "get_env"  # 获取环境信息
API_TYPE_SYNC_L1_TARGET_CODES = "sync_l1_subscript_codes"  # 同步L1需要订阅的代码
API_TYPE_SYSTEM_LOG = "system_log"  # 系统日志

API_TYPE_GET_FROM_DATA_SERVER = "get_from_data_server"  # 从数据服务器拉取数据

class ActionCallback(object):
    # 交易
@@ -91,6 +91,9 @@
        pass

    def OnSyncL2SubscriptCodes(self, client_id, request_id):
        pass

    def OnGetFromDataServer(self, client_id, request_id, data):
        pass


@@ -190,6 +193,8 @@
                            cls.action_callback.OnSyncL2SubscriptCodes(client_id, request_id)
                        elif content_type == API_TYPE_SYSTEM_LOG:
                            cls.action_callback.OnSystemLog(client_id, request_id, data)
                        elif content_type == API_TYPE_GET_FROM_DATA_SERVER:
                            cls.action_callback.OnGetFromDataServer(client_id, request_id, data)
                    except Exception as e:
                        logging.exception(e)
                        pass

 test/l2_trade_test.py

@@ -12,6 +12,7 @@

from code_attribute import big_money_num_manager, gpcode_manager
from db.redis_manager_delegate import RedisUtils
from l2.huaxin import huaxin_delegate_postion_manager
from log_module import log, log_export, async_log_util
from trade.huaxin import huaxin_trade_api
from utils import tool
@@ -83,7 +84,7 @@
                except Exception as e:
                    pass

    # @unittest.skip("跳过此单元测试")
    @unittest.skip("跳过此单元测试")
    def test_trade(self):
        threading.Thread(target=async_log_util.run_sync,daemon=True).start()
        code = "603178"
@@ -170,6 +171,13 @@
            l2.l2_data_manager_new.L2TradeDataProcessor.process_add_datas(code, total_datas[indexs[0]:indexs[1] + 1], 0,
                                                                          0)

    def test_place_order(self):
        code = "002241"
        l2.l2_data_util.load_l2_data(code)
        total_datas = deepcopy(l2.l2_data_util.local_today_datas[code])
        huaxin_delegate_postion_manager.place_order(code, 17.36, 100, total_datas[753])
        huaxin_delegate_postion_manager.get_l2_place_order_position(code,total_datas[755:1038])

    @unittest.skip("跳过此单元测试")
    def test_h_cancel(self):
        code = "600540"

 trade/huaxin/huaxin_trade_server.py

@@ -16,6 +16,8 @@

import dask
import psutil
import requests

import huaxin_client.constant
from line_profiler import LineProfiler

@@ -508,7 +510,7 @@
                print("发送数据成功")
                break
            except Exception as e1:
                pass
                logging.exception(e1)

    # 交易
    def OnTrade(self, client_id, request_id, data):
@@ -803,6 +805,21 @@
        except Exception as e:
            logger_debug.error(e)

    def OnGetFromDataServer(self, client_id, request_id, data):
        path = data["path"]
        params = data["params"]
        params_strs = []
        if params:
            for k in params:
                params_strs.append(f"{k}={params[k]}")
        if params_strs:
            path += "?"
            path += "&".join(params_strs)
        # 获取参数
        response = requests.get(f"http://127.0.0.1:9004{path}")
        if response.status_code == 200:
            self.send_response(response.text, client_id, request_id)


class MyL2DataCallback(l2_data_transform_protocol.L2DataCallBack):
    def OnL2Order(self, code, datas, timestamp):
@@ -905,4 +922,3 @@
        rate = round(total_deal_nums / (thresh_hold_money // (float(limit_up_price) * 100)), 2)
        LCancelRateManager().set_big_num_deal_rate(code, rate)
        print("撤单比例", LCancelRateManager().get_cancel_rate(code))