quant/big-qmt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add zt_open_hands

Browse Source
This commit is contained in:
yanweidong
2026-09-14 19:11:58 +08:00
parent 9eac1d2e10
commit c8151f4a3f
26 changed files with 126 additions and 550 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
py-client/tests/test_zt_trading.py
View File

@@ -1,14 +1,12 @@
import tempfile
import unittest
from concurrent.futures import ThreadPoolExecutor
from contextlib import closing
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace as NS
from unittest.mock import Mock, patch
from libs.grid_take_profit import GridState
from libs.state import FLAG_BUY, State
from libs.state import State
from sdk import Assets, DealItem, PositionItem, Tick
from strategy.zt import boot
from strategy.zt.positions import manage_positions
@@ -17,7 +15,7 @@ from strategy.zt.positions import manage_positions
class ZTTradingTests(unittest.TestCase):
def setUp(self):
self.code = '600000.SH'
self.run = NS(account_cfg=NS(account_id='test', strategy='zt', buy_value=1000,
self.run = NS(account_cfg=NS(account_id='test', strategy='zt', buy_value=1000, zt_open_hands=1,
excluded_codes=[], enable_loss_add_position=False,
min_cash_ratio=0.1),
orders=Mock(), client=Mock(), profit_tracker=Mock(), add_watch=Mock())
@@ -66,44 +64,6 @@ class ZTTradingTests(unittest.TestCase):
error.assert_not_called()
self.run.orders.place.assert_not_called()
def test_run_once_quarantines_manual_trade_but_manages_good_stock(self):
with tempfile.TemporaryDirectory() as tmp, ThreadPoolExecutor(max_workers=2) as executor:
store = State(Path(tmp) / 'state.db')
good = '600001.SH'
positions = [PositionItem(stock_code=c, volume=100, can_use_volume=100, open_price=10)
for c in [self.code, good]]
store.sync_account(positions, [], initialize=True)
manual = DealItem(stock_code=self.code, order_sys_id='manual', remark='',
offset_flag=FLAG_BUY, volume=100, price=10, trade_amount=1000)
self.run.executor = executor
self.run.client.deals.return_value = [manual]
self.run.client.portfolio.return_value = NS(
assets=Assets(10000, 10000), positions={p.stock_code: p for p in positions}, orders=[])
self.run.client.full_tick.return_value = {p.stock_code: Tick(last_price=11) for p in positions}
with patch.object(boot, 'datetime') as clock, patch.object(boot, 'market_allow_open', return_value=True):
clock.now.return_value = datetime(2026, 9, 11, 10)
boot.RunOnce(self.run, store, [])
self.assertEqual(store.blocked_codes, {self.code})
self.run.orders.refresh.assert_called_once()
self.assertEqual(self.run.orders.place.call_count, 1)
self.assertEqual(self.run.orders.place.call_args.args[1].code, good)
def test_run_once_does_not_reopen_quarantined_sold_out_code(self):
with tempfile.TemporaryDirectory() as tmp, ThreadPoolExecutor(max_workers=2) as executor:
store = State(Path(tmp) / 'state.db')
store.sync_account([], [], initialize=True)
self.run.executor = executor
self.run.client.deals.return_value = [DealItem(
stock_code=self.code, order_sys_id='manual', remark='', offset_flag=FLAG_BUY,
volume=100, price=10, trade_amount=1000)]
self.run.client.portfolio.return_value = NS(assets=Assets(10000, 10000), positions={}, orders=[])
self.run.client.full_tick.return_value = {}
with patch.object(boot, 'datetime') as clock, patch.object(boot, 'market_allow_open', return_value=True), \
patch.object(boot, 'open_signal') as opened:
clock.now.return_value = datetime(2026, 9, 11, 10)
boot.RunOnce(self.run, store, [NS(code=self.code)])
opened.assert_not_called()
def start(self, client, directory):
self.run.account_cfg.grid_step_pct = 1
global_cfg = NS(qmt_base_url='unused', qmt_token='', qmt_data_dir=directory)