L1测试

Administrator

2024-02-02 1aaa8a3ef115b3b4d535f4668bd2fd3701da2cee

L1测试

 huaxin_client/l1_client_for_trade.py

@@ -8,7 +8,7 @@
from huaxin_client import socket_util
import xmdapi
from huaxin_client import constant
from log_module.log import logger_system, logger_local_huaxin_l1
from log_module.log import logger_system, logger_local_huaxin_l1, logger_local_huaxin_l1_trade_info

################B类##################
ADDRESS = "udp://224.224.1.19:7880"
@@ -129,10 +129,12 @@
        if pMarketDataField.SecurityName.find("ST") >= 0:
            return
        rate = 0
        self.l1_data_queue.append((
        item = (
            pMarketDataField.SecurityID, pMarketDataField.LastPrice, rate, pMarketDataField.Volume,
            pMarketDataField.UpdateTime,
            pMarketDataField.BidPrice1, pMarketDataField.BidVolume1))
            pMarketDataField.BidPrice1, pMarketDataField.BidVolume1)
        self.l1_data_queue.append(item)
        logger_local_huaxin_l1_trade_info.info(f"获取到L1数据：{item}")

        # print(
        #     "SecurityID[%s] SecurityName[%s] LastPrice[%.2f] Volume[%d] Turnover[%.2f] BidPrice1[%.2f] BidVolume1[%d] AskPrice1[%.2f] AskVolume1[%d] UpperLimitPrice[%.2f] LowerLimitPrice[%.2f]"
@@ -233,7 +235,7 @@
    def test_sub():
        time.sleep(5)
        queue_l1_trade_r_strategy_w.put_nowait(
            json.dumps({"type": "set_target_codes", "data": ["603825", "603767", "603778"]}))
            json.dumps({"type": "set_target_codes", "data": ["603990"]}))

    def read_data():
        while True:

 test_l1.py

New file
@@ -0,0 +1,22 @@
import json
import multiprocessing
import threading
import time

from huaxin_client import l1_client_for_trade

if __name__ == "__main__":
    def test_sub():
        time.sleep(10)
        queue_l1_trade_r_strategy_w.put_nowait(
            json.dumps({"type": "set_target_codes", "data": ["603990"]}))

    def read_data():
        while True:
            val = queue_l1_trade_w_strategy_r.get()
            print(val)

    queue_l1_trade_w_strategy_r, queue_l1_trade_r_strategy_w = multiprocessing.Queue(), multiprocessing.Queue()
    threading.Thread(target=test_sub, daemon=True).start()
    threading.Thread(target=read_data, daemon=True).start()
    l1_client_for_trade.run(queue_l1_trade_w_strategy_r, queue_l1_trade_r_strategy_w)

 test_l1.spec

New file
@@ -0,0 +1,50 @@
# -*- mode: python ; coding: utf-8 -*-


block_cipher = None


a = Analysis(
    ['test_l1.py'],
    pathex=[],
    binaries=[],
    datas=[],
    hiddenimports=[],
    hookspath=[],
    hooksconfig={},
    runtime_hooks=[],
    excludes=[],
    win_no_prefer_redirects=False,
    win_private_assemblies=False,
    noarchive=False,
    cipher=block_cipher,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)

exe = EXE(
    pyz,
    a.scripts,
    [],
    exclude_binaries=True,
    name='gp_server_l1_test',
    debug=False,
    bootloader_ignore_signals=False,
    strip=False,
    upx=True,
    console=True,
    disable_windowed_traceback=False,
    argv_emulation=False,
    target_arch=None,
    codesign_identity=None,
    entitlements_file=None,
)
coll = COLLECT(
    exe,
    a.binaries,
    a.zipfiles,
    a.datas,
    strip=False,
    upx=True,
    upx_exclude=[],
    name='test_l1',
)