79 lines
3.8 KiB
Python
79 lines
3.8 KiB
Python
import ast
|
||
import unittest
|
||
from dataclasses import asdict, fields
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
from unittest.mock import Mock
|
||
from libs.order import OrderBook as ActiveOrders
|
||
from sdk.models import Assets, DealItem, OrderItem, PositionItem
|
||
from sdk.portfolio import PortfolioMixin
|
||
|
||
# QMT 委托/成交的 offset_flag：48 买入、49 卖出。
|
||
FLAG_BUY = 48
|
||
|
||
|
||
class ApiModelTests(unittest.TestCase):
|
||
def setUp(self):
|
||
source = Path(__file__).resolve().parents[2] / 'api' / 'qmt_rest_new.py'
|
||
names = {'format_assets', 'format_holding', 'format_orders', 'format_deals'}
|
||
nodes = [n for n in ast.parse(source.read_text(encoding='utf-8')).body
|
||
if isinstance(n, ast.FunctionDef) and n.name in names]
|
||
ns = {'HTTPError': RuntimeError}
|
||
exec(compile(ast.Module(body=nodes, type_ignores=[]), str(source), 'exec'), ns)
|
||
attrs = {n.attr: '' if n.attr.startswith('m_str') else 0
|
||
for node in nodes for n in ast.walk(node)
|
||
if isinstance(n, ast.Attribute) and n.attr.startswith('m_')}
|
||
attrs.update(m_strInstrumentID='600000', m_strExchangeID='SH',
|
||
m_strOrderSysID='sys1', m_strRemark='trend-BUY-1|trend',
|
||
m_nOffsetFlag=FLAG_BUY, m_nOrderStatus=56, m_nVolume=100,
|
||
m_nVolumeTraded=100, m_nVolumeTotalOriginal=100,
|
||
m_dPrice=10.0, m_dTradeAmount=1000.0, m_dBalance=2000.0,
|
||
m_dAvailable=1000.0, m_strInsertDate='20260907',
|
||
m_strInsertTime='100000', m_strTradeDate='20260907', m_strTradeTime='100000')
|
||
obj = SimpleNamespace(**attrs)
|
||
self.assets = ns['format_assets']([obj])
|
||
self.positions = ns['format_holding']([obj])
|
||
self.orders = ns['format_orders']([obj])
|
||
self.deals = ns['format_deals']([obj])
|
||
self.client = PortfolioMixin()
|
||
self.client._get_json = {
|
||
'/api/portfolio/assets': self.assets, '/api/portfolio/positions': self.positions,
|
||
'/api/portfolio/order': self.orders, '/api/portfolio/deal': self.deals,
|
||
'/api/portfolio': {'assets': self.assets, 'positions': self.positions, 'orders': self.orders},
|
||
}.__getitem__
|
||
|
||
def test_models_exactly_match_api_keys_and_values(self):
|
||
for model, row in ((Assets, self.assets), (PositionItem, self.positions['600000.SH']),
|
||
(OrderItem, self.orders[0]), (DealItem, self.deals[0])):
|
||
self.assertEqual({field.name for field in fields(model)}, set(row))
|
||
self.assertEqual(asdict(model(**row)), row)
|
||
|
||
def test_all_endpoints(self):
|
||
self.assertEqual(asdict(self.client.assets()), self.assets)
|
||
codes, positions = self.client.positions()
|
||
self.assertEqual(codes, ['600000.SH'])
|
||
self.assertEqual(asdict(positions[0]), self.positions[codes[0]])
|
||
self.assertEqual(asdict(self.client.orders()[0]), self.orders[0])
|
||
self.assertEqual(asdict(self.client.deals()[0]), self.deals[0])
|
||
portfolio = self.client.portfolio()
|
||
self.assertEqual(asdict(portfolio.positions[codes[0]]), self.positions[codes[0]])
|
||
self.assertEqual(asdict(portfolio.orders[0]), self.orders[0])
|
||
|
||
def test_derived_properties_and_order_cache(self):
|
||
order = self.client.orders()[0]
|
||
self.assertEqual(order.side, 'BUY')
|
||
self.assertEqual(order.local_order_id, 'trend-BUY-1')
|
||
self.assertEqual(order.created_at, datetime(2026, 9, 7, 10))
|
||
order.order_status = 50
|
||
order.insert_date = '20000101'
|
||
client = Mock()
|
||
book = ActiveOrders()
|
||
book.refresh(client, [order])
|
||
client.cancel_by_id.assert_called_once_with('sys1')
|
||
self.assertTrue(book.busy('600000.SH', 'BUY'))
|
||
|
||
|
||
if __name__ == '__main__':
|
||
unittest.main()
|