92 lines
4.0 KiB
Python
92 lines
4.0 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace as NS
|
|
from unittest.mock import Mock, patch
|
|
|
|
import yaml
|
|
|
|
import config
|
|
from config import AccountConfig, GlobalConfig, SignalConfig
|
|
from sdk import Tick
|
|
from strategy.zt import boot
|
|
from strategy.zt.open import open_signal
|
|
from strategy.zt.positions import handle_loss
|
|
|
|
|
|
class ZTOpenHandsTests(unittest.TestCase):
|
|
def runtime(self, hands):
|
|
run = NS(
|
|
account_cfg=AccountConfig(zt_open_hands=hands, buy_value=10000, strategy='zt'),
|
|
global_cfg=GlobalConfig(signals={'dcm': SignalConfig()}),
|
|
client=Mock(), orders=Mock(), open_watch=Mock(), add_watch=Mock(),
|
|
)
|
|
run.orders.busy.return_value = False
|
|
run.orders.place.return_value = True
|
|
run.open_watch.triggered.return_value = True
|
|
run.add_watch.triggered.return_value = True
|
|
return run
|
|
|
|
def test_open_and_add_use_same_hands_at_different_prices(self):
|
|
run = self.runtime(3)
|
|
code = '600000.SH'
|
|
for price in (8, 12):
|
|
with self.subTest(price=price):
|
|
open_signal(run, {code: Tick(last_price=price)},
|
|
[NS(code=code, signal_key='dcm', last_close=10)])
|
|
self.assertEqual(run.orders.place.call_args.args[1].volume, 300)
|
|
decision = handle_loss(run, code, 100, Tick(last_price=price), -20, 5000)
|
|
self.assertTrue(decision.submitted)
|
|
self.assertEqual(run.orders.place.call_args.args[1].volume, 300)
|
|
self.assertEqual(decision.reserved_cash, price * 300)
|
|
|
|
def test_add_does_not_reduce_hands_when_cash_is_insufficient(self):
|
|
run = self.runtime(3)
|
|
decision = handle_loss(run, '600000.SH', 100, Tick(last_price=10), -20, 2999)
|
|
self.assertFalse(decision.submitted)
|
|
run.orders.place.assert_not_called()
|
|
|
|
def test_zero_hands_does_not_initialize_strategy(self):
|
|
with patch.object(config, 'account_config', AccountConfig()), \
|
|
patch.object(boot, 'Client') as client, \
|
|
patch.object(boot, 'State') as state, \
|
|
patch.object(boot, 'ThreadPoolExecutor') as executor, \
|
|
patch.object(boot, 'init_signals') as signals:
|
|
boot.StartZT()
|
|
for dependency in (client, state, executor, signals):
|
|
dependency.assert_not_called()
|
|
|
|
def test_zero_hands_never_submits_open_or_add_orders(self):
|
|
run = self.runtime(0)
|
|
code = '600000.SH'
|
|
open_signal(run, {code: Tick(last_price=10)},
|
|
[NS(code=code, signal_key='dcm', last_close=10)])
|
|
decision = handle_loss(run, code, 100, Tick(last_price=10), -20, 10000)
|
|
self.assertFalse(decision.submitted)
|
|
run.orders.place.assert_not_called()
|
|
|
|
def test_config_accepts_only_nonnegative_integer_hands(self):
|
|
with tempfile.TemporaryDirectory() as directory, \
|
|
patch.object(config, 'global_config'), patch.object(config, 'account_config'):
|
|
root = Path(directory)
|
|
(root / '_global.yaml').write_text(yaml.safe_dump({
|
|
'qmt_base_url': 'unused', 'api_host': 'unused',
|
|
'qmt_data_dir': directory, 'hosts': {'test': 'account'},
|
|
}), encoding='utf-8')
|
|
for hands in (None, 0, 3, -1, 1.5, '3', True):
|
|
with self.subTest(hands=hands):
|
|
account = dict(buy_value=1000, strategy='zt', signal_allow=['dcm'])
|
|
if hands is not None:
|
|
account['zt_open_hands'] = hands
|
|
(root / 'account.yaml').write_text(yaml.safe_dump(account), encoding='utf-8')
|
|
if hands is None or type(hands) is int and hands >= 0:
|
|
_, loaded = config.load(root, 'test')
|
|
self.assertEqual(loaded.zt_open_hands, hands or 0)
|
|
else:
|
|
with self.assertRaisesRegex(ValueError, 'zt_open_hands'):
|
|
config.load(root, 'test')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|