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

fix bug

Browse Source
This commit is contained in:
yanweidong
2026-09-05 11:46:31 +08:00
parent 3689a7ad86
commit 63ffc1329b
13 changed files with 229 additions and 731 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
py-client/tests/test_ipo.py
View File

@@ -1,140 +0,0 @@
from __future__ import annotations
import unittest
from datetime import datetime
from tempfile import TemporaryDirectory
from types import SimpleNamespace
from unittest.mock import patch
from strategy.ipo.boot import AutoBuyIpo
RUN_TIME = datetime(2026, 8, 28, 10, 0)
class FakeClient:
def __init__(self, candidates=None, orders=None, deals=None, fail_codes=None):
self.candidates = candidates or {}
self.orders = orders or []
self.deal_rows = deals or []
self.fail_codes = set(fail_codes or [])
self.submissions = []
self.closed = False
def __enter__(self):
return self
def __exit__(self, *_args):
self.closed = True
def trading_dates(self, *_args):
return ["20260828"]
def trade_detail_data(self, datatype):
self.assert_order_type = datatype
return self.orders
def deals(self):
return self.deal_rows
def ipo_data(self, ipo_type):
self.assert_ipo_type = ipo_type
return self.candidates
def passorder(self, **kwargs):
code = kwargs["stock"]
self.submissions.append(kwargs)
if code in self.fail_codes:
raise RuntimeError("simulated rejection")
return {"status": "success", "order_ref": f"ref-{code}"}
class AutoBuyIpoTests(unittest.TestCase):
def _configs(self, directory, enabled=True):
return (
SimpleNamespace(
qmt_base_url="http://qmt",
qmt_token="token",
qmt_data_dir=directory,
),
SimpleNamespace(account_id="account-A", enable_auto_ipo=enabled),
)
def test_disabled_does_not_create_client(self):
with TemporaryDirectory() as directory:
global_cfg, account_cfg = self._configs(directory, enabled=False)
with (
patch("strategy.ipo.boot.config.global_config", global_cfg),
patch("strategy.ipo.boot.config.account_config", account_cfg),
patch("strategy.ipo.boot.Client") as client_factory,
):
self.assertEqual(AutoBuyIpo(RUN_TIME), 0)
client_factory.assert_not_called()
def test_local_record_prevents_duplicate_after_restart(self):
candidates = {
"688001.SH": {"issuePrice": 10, "maxPurchaseNum": 1000},
}
first = FakeClient(candidates=candidates)
second = FakeClient(candidates=candidates)
with TemporaryDirectory() as directory:
global_cfg, account_cfg = self._configs(directory)
with (
patch("strategy.ipo.boot.config.global_config", global_cfg),
patch("strategy.ipo.boot.config.account_config", account_cfg),
patch("strategy.ipo.boot.Client", side_effect=[first, second]),
):
self.assertEqual(AutoBuyIpo(RUN_TIME), 1)
self.assertEqual(AutoBuyIpo(RUN_TIME), 0)
self.assertEqual(len(first.submissions), 1)
self.assertEqual(second.submissions, [])
self.assertTrue(first.closed)
self.assertTrue(second.closed)
def test_broker_order_prevents_duplicate(self):
candidates = {
"688001.SH": {"issuePrice": 10, "maxPurchaseNum": 1000},
}
client = FakeClient(
candidates=candidates,
orders=[{
"m_strInstrumentID": "688001",
"m_strInsertDate": "20260828",
"m_strRemark": "IPO_SUBSCRIBE",
}],
)
with TemporaryDirectory() as directory:
global_cfg, account_cfg = self._configs(directory)
with (
patch("strategy.ipo.boot.config.global_config", global_cfg),
patch("strategy.ipo.boot.config.account_config", account_cfg),
patch("strategy.ipo.boot.Client", return_value=client),
):
self.assertEqual(AutoBuyIpo(RUN_TIME), 0)
self.assertEqual(client.submissions, [])
def test_one_rejection_does_not_stop_other_candidates(self):
candidates = {
"688001.SH": {"issuePrice": 10, "maxPurchaseNum": 1000},
"688002.SH": {"issuePrice": 20, "maxPurchaseNum": 500},
}
client = FakeClient(candidates=candidates, fail_codes={"688001.SH"})
with TemporaryDirectory() as directory:
global_cfg, account_cfg = self._configs(directory)
with (
patch("strategy.ipo.boot.config.global_config", global_cfg),
patch("strategy.ipo.boot.config.account_config", account_cfg),
patch("strategy.ipo.boot.Client", return_value=client),
):
self.assertEqual(AutoBuyIpo(RUN_TIME), 1)
self.assertEqual(
[item["stock"] for item in client.submissions],
["688001.SH", "688002.SH"],
)
self.assertTrue(client.closed)
if __name__ == "__main__":
unittest.main()

22
py-client/tests/test_market.py
View File

@@ -1,22 +0,0 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from libs.market import market_allow_open, refresh_market
class MarketCacheTests(unittest.TestCase):
def test_refresh_updates_open_cache(self):
with patch("libs.market.get_json", return_value={"data": {"action": "UP"}}):
self.assertEqual(refresh_market("http://example"), "UP")
self.assertTrue(market_allow_open())
def test_refresh_failure_blocks_open(self):
with patch("libs.market.get_json", side_effect=OSError("offline")):
self.assertEqual(refresh_market("http://example"), "UNKNOWN")
self.assertFalse(market_allow_open())
if __name__ == "__main__":
unittest.main()

33
py-client/tests/test_signal.py
View File

@@ -1,33 +0,0 @@
from types import SimpleNamespace
import unittest
from unittest.mock import patch
from libs.signal import SignalResult, fetch_signal, init_signals
class SignalTests(unittest.TestCase):
def test_fetch_failure_returns_empty_result(self):
with patch("libs.signal.get_json", side_effect=OSError("offline")):
self.assertEqual(fetch_signal("http://example", "/signals"), SignalResult())
def test_init_signals_continues_after_fetch_failure(self):
config = SimpleNamespace(
api_host="http://example",
signals={
"failed": SimpleNamespace(url="/failed"),
"working": SimpleNamespace(url="/working"),
},
)
responses = [
SignalResult(),
SignalResult(data={"000001.SZ": SimpleNamespace(signal_key="")}),
]
with patch("libs.signal.fetch_signal", side_effect=responses):
result = init_signals(config, ["failed", "working"])
self.assertEqual(len(result), 1)
self.assertEqual(result[0].signal_key, "working")
if __name__ == "__main__":
unittest.main()

359
py-client/tests/test_trend.py
View File

@@ -1,359 +0,0 @@
from __future__ import annotations
import unittest
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
from tempfile import TemporaryDirectory
from types import SimpleNamespace
from unittest.mock import patch
from libs.grid_take_profit import GridState, GridTrailingTracker
from sdk import APIError, Assets, OrderItem, Portfolio, PositionItem, Tick
from strategy.trend.order import OrderBook, PlaceOrderRequest
from strategy.trend.open import do_open
from strategy.trend.positions import LOSS_TIERS, handle_loss, manage_positions
from strategy.trend.boot import RunOnce
from strategy.trend.state import STATUS_OK, State, StateItem
class FakeClient:
def __init__(self):
self.orders = []
def passorder_latest_tagged(self, op, code, volume, strategy_name, order_id):
self.orders.append((op, code, volume, strategy_name, order_id))
return {"status": "success", "order_ref": f"broker-{len(self.orders)}"}
class FakeOrderClient:
def __init__(self, orders):
self.orders = orders
self.canceled = []
def trade_detail_data(self, _datatype):
return self.orders
def cancel_by_id(self, order_id):
self.canceled.append(order_id)
class FailedOrderClient:
def passorder_latest_tagged(self, *_args):
raise APIError(502, "QMT did not return a valid order reference")
class TrendTests(unittest.TestCase):
def test_trend_order_id_format(self):
self.assertRegex(OrderBook.new_order_id(), r"^trend-[0-9a-f]{24}$")
def test_open_records_only_pending_order(self):
with TemporaryDirectory() as directory:
state = State.for_strategy(directory, "trend", "A")
forgotten = []
runtime = SimpleNamespace(
client=FakeClient(),
orders=OrderBook(),
state=state,
open_watch=SimpleNamespace(forget=forgotten.append),
)
do_open(runtime, "000001.SZ", 100, "morning", 12.345)
item = state.get("000001.SZ")
self.assertRegex(item.base_order_id, r"^trend-[0-9a-f]{24}$")
self.assertEqual(item.base_qty, 0)
self.assertEqual(item.base_cost, 0)
self.assertEqual(item.base_status, "ING")
self.assertEqual(forgotten, ["000001.SZ"])
def test_pending_base_order_survives_position_delay(self):
with TemporaryDirectory() as directory:
state = State.for_strategy(directory, "trend", "A")
state.set(StateItem(
"000001.SZ",
base_order_id="trend-12345678",
base_status="ING",
))
pending = OrderItem(
"1", "000001.SZ", "BUY", "", "50", None, 100,
local_order_id="trend-12345678",
)
state.reconcile([], [pending])
self.assertEqual(
state.get("000001.SZ").base_order_id,
"trend-12345678",
)
def test_grid_states_and_account_isolation(self):
tracker = GridTrailingTracker(1)
self.assertEqual(tracker.observe("A:code", 2.1).state, GridState.ARMED)
self.assertEqual(tracker.observe("A:code", 3.1).state, GridState.RAISED)
self.assertEqual(tracker.observe("A:code", 2.9).state, GridState.RETREAT)
self.assertEqual(tracker.observe("B:code", 2.9).state, GridState.ARMED)
tracker.retain([])
self.assertEqual(tracker.observe("A:code", 2.9).state, GridState.ARMED)
def test_order_book_locks_duplicate_order(self):
client = FakeClient()
book = OrderBook()
request = PlaceOrderRequest(client, 23, "000001.SZ", 100, "local", "morning")
self.assertTrue(book.place(request))
self.assertTrue(book.busy("000001.SZ", "BUY"))
def test_order_api_error_returns_false_with_traceback(self):
book = OrderBook()
request = PlaceOrderRequest(FailedOrderClient(), 23, "000001.SZ", 100, "local", "morning")
with self.assertLogs(level="ERROR") as captured:
self.assertFalse(book.place(request))
output = "\n".join(captured.output)
self.assertIn("HTTP状态=502", output)
self.assertIn("Traceback", output)
def test_refresh_tracks_active_and_completed_and_cancels_expired(self):
old = datetime.now() - timedelta(seconds=20)
orders = [
OrderItem("active", "A", "BUY", "", "49", old, 100),
OrderItem("completed", "B", "SELL", "", "56", old, 100),
OrderItem("canceled", "C", "BUY", "", "54", old, 100),
OrderItem("failed", "D", "BUY", "", "57", old, 100),
]
client = FakeOrderClient(orders)
book = OrderBook(cancel_timeout_sec=10)
book.refresh(client, orders)
self.assertEqual({item.id for item in book.data}, {"completed"})
self.assertEqual(client.canceled, ["active"])
def test_position_dataclasses_execute_without_type_error(self):
with TemporaryDirectory() as directory:
state = State.for_strategy(directory, "trend", "A")
position = PositionItem(
stock_code="000001.SZ", volume=100, can_use_volume=100,
open_price=10, market_value=1000,
)
state.sync_positions([position])
runtime = SimpleNamespace(
client=FakeClient(), state=state, orders=OrderBook(),
open_watch=SimpleNamespace(forget=lambda _code: None),
add_watch=SimpleNamespace(triggered=lambda *_args: False, forget=lambda _code: None),
profit_tracker=GridTrailingTracker(1),
account_cfg=SimpleNamespace(
account_id="A", excluded_codes=[], grid_step_pct=1,
enable_loss_add_position=False, buy_value=5000,
strategy="trend",
),
)
manage_positions(runtime, {"000001.SZ": Tick(last_price=10.1)}, [position], True, 5000)
def test_position_log_contains_code_name_profit_and_loss_actions(self):
with TemporaryDirectory() as directory:
state = State.for_strategy(directory, "trend", "A")
position = PositionItem(
stock_code="000001.SZ", stock_name="平安银行", volume=100,
can_use_volume=100, open_price=10, market_value=1000,
)
state.sync_positions([position])
runtime = SimpleNamespace(
client=FakeClient(), state=state, orders=OrderBook(),
add_watch=SimpleNamespace(triggered=lambda *_args: False),
profit_tracker=GridTrailingTracker(1),
account_cfg=SimpleNamespace(
account_id="A", excluded_codes=[], grid_step_pct=1,
enable_loss_add_position=False, buy_value=5000,
strategy="trend",
),
)
with self.assertLogs(level="INFO") as captured:
manage_positions(
runtime,
{"000001.SZ": Tick(last_price=10.1)},
[position],
True,
5000,
)
output = "\n".join(captured.output)
self.assertIn("代码=000001.SZ", output)
self.assertIn("名称=平安银行", output)
self.assertIn("止盈=未触发", output)
self.assertIn("补仓=未启用", output)
def test_loss_tier_boundary_does_not_overflow(self):
self.assertEqual(len(LOSS_TIERS), 2)
with TemporaryDirectory() as directory:
state = State.for_strategy(directory, "trend", "A")
position = PositionItem(stock_code="A", volume=100, open_price=10, market_value=1000)
state.sync_positions([position])
item = state.get("A")
item.added_num = len(LOSS_TIERS)
state.set(item)
runtime = SimpleNamespace(
state=state, account_cfg=SimpleNamespace(buy_value=5000, strategy="trend"),
add_watch=SimpleNamespace(triggered=lambda *_args: True), orders=OrderBook(),
client=FakeClient(),
)
decision = handle_loss(runtime, position, Tick(last_price=5), -60, 5000)
self.assertFalse(decision.submitted)
def test_loss_tiers_zero_and_one(self):
with TemporaryDirectory() as directory:
state = State.for_strategy(directory, "trend", "A")
position = PositionItem(stock_code="A", volume=100, open_price=10, market_value=1000)
state.sync_positions([position])
runtime = SimpleNamespace(
state=state, account_cfg=SimpleNamespace(buy_value=5000, strategy="trend"),
add_watch=SimpleNamespace(triggered=lambda *_args: False),
orders=OrderBook(), client=FakeClient(),
)
first = handle_loss(runtime, position, Tick(last_price=7), -30, 5000)
self.assertIn("等待", first.message)
item = state.get("A")
item.added_num = 1
state.set(item)
before_second_tier = handle_loss(runtime, position, Tick(last_price=6), -40, 5000)
self.assertEqual(before_second_tier.message, "")
second = handle_loss(runtime, position, Tick(last_price=5), -50, 5000)
self.assertIn("等待", second.message)
def test_reconcile_split_orders_complete_only_when_all_are_status_56(self):
with TemporaryDirectory() as directory:
state = State.for_strategy(directory, "trend", "A")
position = PositionItem(stock_code="A", volume=100, open_price=10)
state.set(StateItem("A", base_order_id="local-1", base_status="ING"))
completed = OrderItem(
"1", "A", "BUY", "", "56", None, 50, "local-1",
traded_volume=50, trade_price=10.1,
)
processing = OrderItem("2", "A", "BUY", "", "50", None, 50, "local-1")
state.reconcile([position], [completed, processing])
self.assertEqual(state.get("A").base_status, "ING")
state.reconcile(
[position],
[
completed,
OrderItem(
"2", "A", "BUY", "", "56", None, 50, "local-1",
traded_volume=50, trade_price=10.3,
),
],
)
item = state.get("A")
self.assertEqual(item.base_status, STATUS_OK)
self.assertEqual(item.base_qty, 100)
self.assertEqual(item.base_cost, 10.2)
canceled = OrderItem("2", "A", "BUY", "", "54", None, 50, "local-1")
item = state.get("A")
item.base_status = "ING"
state.set(item)
state.reconcile([position], [completed, canceled])
self.assertEqual(state.get("A").base_status, "")
def test_reconcile_records_filled_add_order(self):
with TemporaryDirectory() as directory:
state = State.for_strategy(directory, "trend", "A")
position = PositionItem(stock_code="A", volume=200, open_price=10)
state.set(StateItem(
"A",
base_qty=100,
base_cost=10,
base_status=STATUS_OK,
added_order_id="add-1",
added_status="ING",
))
completed = OrderItem(
"1", "A", "BUY", "", "56", None, 100, "add-1",
traded_volume=100, trade_amount=950,
)
state.reconcile([position], [completed])
item = state.get("A")
self.assertEqual(item.added_status, STATUS_OK)
self.assertEqual(item.added_num, 1)
self.assertEqual(item.added_qty, 100)
self.assertEqual(item.added_cost, 9.5)
def test_low_cash_still_runs_position_management(self):
client = SimpleNamespace(
portfolio=lambda: Portfolio(
assets=Assets(total=10000, available=10),
positions={"A": PositionItem(stock_code="A", volume=100, open_price=10)},
orders=[],
),
full_tick=lambda _codes: {"A": Tick(last_price=11)},
)
with ThreadPoolExecutor(max_workers=2) as executor:
runtime = SimpleNamespace(
client=client,
account_cfg=SimpleNamespace(min_cash_ratio=0.1),
global_cfg=SimpleNamespace(api_host="http://example"),
orders=SimpleNamespace(refresh=lambda _client, _orders: None, data=[]),
state=SimpleNamespace(
codes=["A"],
reconcile=lambda *_args: None,
),
executor=executor,
)
with (
patch("strategy.trend.boot.trading_time", return_value=True),
patch("strategy.trend.boot.market_allow_open", return_value=True),
patch("strategy.trend.boot.open_signal") as open_mock,
patch("strategy.trend.boot.manage_positions") as manage_mock,
):
RunOnce(runtime, [])
open_mock.assert_not_called()
manage_mock.assert_called_once()
def test_state_without_broker_order_allows_reopen(self):
with TemporaryDirectory() as directory:
state = State.for_strategy(directory, "trend", "A")
state.set(StateItem(
"A",
base_order_id="missing-order",
base_qty=100,
base_status="ING",
))
state.save()
client = SimpleNamespace(
portfolio=lambda: Portfolio(
assets=Assets(total=10000, available=5000),
positions={},
orders=[],
),
full_tick=lambda _codes: {"A": Tick(last_price=10)},
)
signal = SimpleNamespace(code="A", signal_key="morning")
with ThreadPoolExecutor(max_workers=2) as executor:
runtime = SimpleNamespace(
client=client,
account_cfg=SimpleNamespace(min_cash_ratio=0.1),
global_cfg=SimpleNamespace(api_host="http://example"),
orders=OrderBook(),
state=state,
open_watch=SimpleNamespace(forget=lambda _code: None),
add_watch=SimpleNamespace(forget=lambda _code: None),
executor=executor,
)
with (
patch("strategy.trend.boot.trading_time", return_value=True),
patch("strategy.trend.boot.market_allow_open", return_value=True),
patch("strategy.trend.boot.open_signal") as open_mock,
patch("strategy.trend.boot.manage_positions"),
):
RunOnce(runtime, [signal])
open_mock.assert_called_once()
self.assertEqual(state.codes, [])
if __name__ == "__main__":
unittest.main()

42
py-client/tests/test_trend_busy.py
View File

@@ -1,42 +0,0 @@
import unittest
from datetime import datetime, timedelta
from unittest.mock import Mock
from sdk import OrderItem
from strategy.trend.order import OrderBook, PlaceOrderRequest
class TrendBusyTests(unittest.TestCase):
def test_snapshot_blocks_order_without_local_cache(self):
book = OrderBook()
client = Mock()
order = OrderItem("1", "A", "BUY", "", "50", None, 100)
book.refresh(client, [order])
self.assertTrue(book.busy("A", "BUY"))
self.assertFalse(book.busy("A", "SELL"))
self.assertFalse(book.place(PlaceOrderRequest(client, 23, "A", 100, "local", "trend")))
client.passorder.assert_not_called()
book.refresh(client, [])
self.assertFalse(book.busy("A", "BUY"))
def test_cancel_request_keeps_order_busy_until_terminal_snapshot(self):
book = OrderBook()
client = Mock()
order = OrderItem("1", "A", "BUY", "", "50", datetime.now() - timedelta(seconds=20), 100)
book.refresh(client, [order])
client.cancel_by_id.assert_called_once_with("1")
self.assertTrue(book.busy("A", "BUY"))
order.status = "54"
book.refresh(client, [order])
self.assertFalse(book.busy("A", "BUY"))
def test_empty_snapshot_keeps_local_cache_protection(self):
book = OrderBook()
client = Mock()
client.passorder.return_value = {"status": "success", "order_ref": "1"}
request = PlaceOrderRequest(client, 23, "A", 100, "local", "trend")
self.assertTrue(book.place(request))
book.refresh(client, [])
self.assertTrue(book.busy("A", "BUY"))
self.assertFalse(book.place(request))
client.passorder.assert_called_once()

35
py-client/tests/test_zt.py
View File

@@ -1,35 +0,0 @@
from __future__ import annotations
import unittest
from datetime import datetime
from tempfile import TemporaryDirectory
from sdk import OrderItem, PositionItem
from strategy.zt.state import BUYING, DONE, SELLING, SOLD, TState, TStateItem
class ZTStateTests(unittest.TestCase):
def test_reconcile_marks_sell_and_buy_orders_completed(self):
with TemporaryDirectory() as directory:
state = TState.for_strategy(directory, "zt", "A")
state.set(TStateItem("A", 1000, 10, "2026-08-31", SELLING, "sell-1", 500, 11))
position = PositionItem(stock_code="A", volume=500, open_price=10)
state.reconcile([position], [OrderItem("1", "A", "SELL", "", "56", datetime.now(), 500, "sell-1")], "2026-08-31")
self.assertEqual(state.get("A").phase, SOLD)
item = state.get("A")
item.phase, item.buy_order_id = BUYING, "buy-1"
state.set(item)
state.reconcile([position], [OrderItem("2", "A", "BUY", "", "56", datetime.now(), 500, "buy-1")], "2026-08-31")
self.assertEqual(state.get("A").phase, DONE)
def test_new_position_becomes_dcm_base_state(self):
with TemporaryDirectory() as directory:
state = TState.for_strategy(directory, "zt", "A")
state.reconcile([PositionItem(stock_code="A", volume=800, open_price=12.5)], [], "2026-08-31")
item = state.get("A")
self.assertEqual((item.base_qty, item.base_cost), (800, 12.5))
if __name__ == "__main__":
unittest.main()