114 lines
5.4 KiB
Python
114 lines
5.4 KiB
Python
"""Offline regression checks for the Python 3.14 performance changes."""
|
|
|
|
import importlib
|
|
import inspect
|
|
import unittest
|
|
from annotationlib import Format, get_annotations
|
|
from concurrent.futures import Future
|
|
from datetime import datetime, time, timedelta
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock, patch
|
|
|
|
from libs.calc import trading_time
|
|
from libs.signal import SignalItem
|
|
from sdk.models import OrderItem, _parse_datetime
|
|
from strategy.trend import boot
|
|
from strategy.trend.open import _parse_minutes, check_timezone
|
|
|
|
|
|
class PerformanceRegressionTests(unittest.TestCase):
|
|
def test_trading_time_matches_original_for_week_and_boundaries(self):
|
|
start = datetime(2026, 9, 7)
|
|
for minute in range(7 * 24 * 60):
|
|
now = start + timedelta(minutes=minute)
|
|
expected = now.weekday() < 5 and (
|
|
time(9, 30) <= now.time() <= time(11, 30)
|
|
or time(13) <= now.time() <= time(15)
|
|
)
|
|
self.assertEqual(trading_time(now), expected, now)
|
|
for clock in ((11, 30), (15, 0)):
|
|
self.assertFalse(trading_time(start.replace(hour=clock[0], minute=clock[1], microsecond=1)))
|
|
|
|
def test_date_parser_preserves_strptime_acceptance(self):
|
|
for date in ('20260907', '2026-09-07', '', '20260229', '20240229', '202691'):
|
|
for clock in ('100000', '10:00:00', '93000', '', '240000', 'bad', '1'):
|
|
try:
|
|
expected = datetime.strptime(date.replace('-', '') + clock.replace(':', '').zfill(6), '%Y%m%d%H%M%S')
|
|
except ValueError:
|
|
expected = None
|
|
self.assertEqual(_parse_datetime(date, clock), expected, (date, clock))
|
|
|
|
def test_date_cache_tracks_mutable_order_fields(self):
|
|
order = OrderItem(insert_date='20260907', insert_time='100000', remark='first|trend')
|
|
self.assertEqual(order.created_at, datetime(2026, 9, 7, 10))
|
|
order.insert_time = '110000'
|
|
order.remark = 'second|trend'
|
|
self.assertEqual(order.created_at, datetime(2026, 9, 7, 11))
|
|
self.assertEqual(order.local_order_id, 'second')
|
|
self.assertEqual(order.get_local_order_id, 'second')
|
|
|
|
def test_caches_are_bounded(self):
|
|
_parse_minutes.cache_clear()
|
|
_parse_datetime.cache_clear()
|
|
for i in range(4200):
|
|
_parse_datetime('invalid', str(i))
|
|
_parse_minutes(str(i))
|
|
self.assertLessEqual(_parse_datetime.cache_info().currsize, 4096)
|
|
self.assertLessEqual(_parse_minutes.cache_info().currsize, 256)
|
|
|
|
def test_timezone_boundaries_and_current_time_not_cached(self):
|
|
for hour in range(24):
|
|
for minute in range(60):
|
|
now = datetime(2026, 9, 7, hour, minute)
|
|
m = hour * 60 + minute
|
|
self.assertEqual(check_timezone('9:30-10:30,invalid,23:00-1:00', now),
|
|
570 <= m <= 630 or m >= 1380 or m <= 60)
|
|
self.assertTrue(check_timezone('*'))
|
|
self.assertFalse(check_timezone('24:00-25:00'))
|
|
self.assertFalse(check_timezone(''))
|
|
|
|
def test_signal_order_duplicates_and_request_order_preserved(self):
|
|
future = Future()
|
|
future.set_result(None)
|
|
assets = SimpleNamespace(available=100, total=100)
|
|
portfolio = SimpleNamespace(assets=assets, positions={'held': object()}, orders=[])
|
|
run = SimpleNamespace(client=Mock(), orders=Mock(), executor=Mock(),
|
|
account_cfg=SimpleNamespace(account_id='test', min_cash_ratio=0.1))
|
|
run.client.portfolio.return_value = portfolio
|
|
run.client.full_tick.return_value = {}
|
|
run.executor.submit.return_value = future
|
|
signals = [SignalItem(code=c) for c in ('new-b', 'held', 'new-a', 'new-b')]
|
|
with patch.object(boot, 'trading_time', return_value=True), \
|
|
patch.object(boot, 'market_allow_open', return_value=True), \
|
|
patch.object(boot, '_cache_portfolio'), patch('builtins.print'):
|
|
boot.RunOnce(run, signals)
|
|
run.client.full_tick.assert_called_once_with(['held', 'new-b', 'new-a'])
|
|
self.assertEqual(run.executor.submit.call_args_list[1].args,
|
|
(boot.open_signal, run, {}, [signals[0], signals[2], signals[3]]))
|
|
|
|
def test_native_annotations_resolve_for_all_application_modules(self):
|
|
root = Path(__file__).resolve().parents[1]
|
|
paths = [p for name in ('config', 'sdk', 'libs', 'strategy')
|
|
for p in (root / name).rglob('*.py')]
|
|
for path in paths:
|
|
parts = list(path.relative_to(root).with_suffix('').parts)
|
|
if parts[-1] == '__init__':
|
|
parts.pop()
|
|
module = importlib.import_module('.'.join(parts))
|
|
for obj in vars(module).values():
|
|
if (inspect.isclass(obj) or inspect.isfunction(obj)) and obj.__module__ == module.__name__:
|
|
get_annotations(obj, format=Format.VALUE)
|
|
if inspect.isclass(obj):
|
|
for member in vars(obj).values():
|
|
if isinstance(member, (classmethod, staticmethod)):
|
|
member = member.__func__
|
|
elif isinstance(member, property):
|
|
member = member.fget
|
|
if inspect.isfunction(member):
|
|
get_annotations(member, format=Format.VALUE)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|