688 lines
23 KiB
Python
688 lines
23 KiB
Python
import base64
|
|
import copy
|
|
import contextlib
|
|
import datetime
|
|
import io
|
|
import json
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
from platforms import joinquant_strategy, qmt_builtin_strategy
|
|
from platforms.fake_joinquant_harness import (
|
|
FakeJoinQuantHarness,
|
|
FakePosition,
|
|
)
|
|
from platforms.fake_qmt_harness import FakeQmtContext, FakeQmtHarness
|
|
from quant60 import portable_core
|
|
from quant60.portable_core import target_package_tape_sha256
|
|
from quant60.target_package import seal_target_package
|
|
|
|
|
|
def _package(
|
|
signal_as_of="2026-07-17T15:00:00+08:00",
|
|
next_session="2026-07-20",
|
|
package_id="TP-20260717-A",
|
|
decision_id="D-20260717-A",
|
|
):
|
|
return seal_target_package(
|
|
{
|
|
"schema_version": "1.0",
|
|
"package_type": "TargetPackageV1",
|
|
"package_id": package_id,
|
|
"decision_id": decision_id,
|
|
"release": "quant60-baseline-60-v1",
|
|
"evidence_class": "provider-snapshot-local",
|
|
"model_id": "ridge-alpha-v1",
|
|
"signal_as_of": signal_as_of,
|
|
"next_session": next_session,
|
|
"model_sha256": "1" * 64,
|
|
"data_sha256": "2" * 64,
|
|
"feature_sha256": "3" * 64,
|
|
"risk_sha256": "4" * 64,
|
|
"cost_sha256": "5" * 64,
|
|
"optimizer_sha256": "6" * 64,
|
|
"config_sha256": "7" * 64,
|
|
"source_sha256": "8" * 64,
|
|
"target_weights": {
|
|
"000001.XSHE": 0.30,
|
|
"600000.XSHG": 0.0,
|
|
},
|
|
"universe": {
|
|
"000001.XSHE": {
|
|
"state": "openable",
|
|
"reasons": ["ELIGIBLE"],
|
|
},
|
|
"600000.XSHG": {
|
|
"state": "sell_only",
|
|
"reasons": ["RISK_REDUCTION"],
|
|
},
|
|
},
|
|
"constraints": {
|
|
"constraint_state": "FEASIBLE",
|
|
"long_only": True,
|
|
"max_single_weight": 0.40,
|
|
"max_gross_weight": 0.80,
|
|
"max_one_way_turnover": 0.80,
|
|
"actual_gross_weight": 0.30,
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
def _jq_histories():
|
|
return {
|
|
"000001.XSHE": [
|
|
10.0 + 0.1 * index for index in range(21)
|
|
],
|
|
"600000.XSHG": [
|
|
15.0 - 0.02 * index for index in range(21)
|
|
],
|
|
}
|
|
|
|
|
|
def _qmt_histories():
|
|
return {
|
|
symbol.replace(".XSHE", ".SZ").replace(".XSHG", ".SH"): values
|
|
for symbol, values in _jq_histories().items()
|
|
}
|
|
|
|
|
|
def _target_config(base, packages):
|
|
config = dict(base)
|
|
config.update(
|
|
{
|
|
"decision_mode": "target_package",
|
|
"target_package_count": len(packages),
|
|
"target_package_tape_sha256": target_package_tape_sha256(
|
|
packages
|
|
),
|
|
"universe_mode": "fixed",
|
|
"universe": sorted(
|
|
{
|
|
symbol
|
|
for package in packages
|
|
for symbol in package["universe"]
|
|
}
|
|
),
|
|
}
|
|
)
|
|
return config
|
|
|
|
|
|
def _evidence_records(output):
|
|
prefix = joinquant_strategy.EVIDENCE_PREFIX
|
|
records = []
|
|
for line in output.getvalue().splitlines():
|
|
if not line.startswith(prefix):
|
|
continue
|
|
encoded = line[len(prefix) :]
|
|
record = json.loads(encoded)
|
|
canonical = json.dumps(
|
|
record,
|
|
ensure_ascii=True,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
allow_nan=False,
|
|
)
|
|
if canonical != encoded:
|
|
raise AssertionError("JoinQuant evidence is not canonical JSON")
|
|
records.append(record)
|
|
return records
|
|
|
|
|
|
class PlatformTargetPackageExecutionTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.jq_config = joinquant_strategy.CONFIG
|
|
self.jq_packages = joinquant_strategy.TARGET_PACKAGES
|
|
self.qmt_config = qmt_builtin_strategy.CONFIG
|
|
self.qmt_packages = qmt_builtin_strategy.TARGET_PACKAGES
|
|
|
|
def tearDown(self):
|
|
joinquant_strategy.CONFIG = self.jq_config
|
|
joinquant_strategy.TARGET_PACKAGES = self.jq_packages
|
|
qmt_builtin_strategy.CONFIG = self.qmt_config
|
|
qmt_builtin_strategy.TARGET_PACKAGES = self.qmt_packages
|
|
|
|
def _install_package(self, package):
|
|
self._install_packages([package])
|
|
|
|
def _install_packages(self, packages):
|
|
packages = copy.deepcopy(packages)
|
|
joinquant_strategy.TARGET_PACKAGES = copy.deepcopy(packages)
|
|
joinquant_strategy.CONFIG = _target_config(
|
|
self.jq_config,
|
|
packages,
|
|
)
|
|
qmt_builtin_strategy.TARGET_PACKAGES = copy.deepcopy(packages)
|
|
qmt_config = _target_config(self.qmt_config, packages)
|
|
qmt_config["universe"] = [
|
|
symbol.replace(".XSHE", ".SZ").replace(".XSHG", ".SH")
|
|
for symbol in qmt_config["universe"]
|
|
]
|
|
qmt_builtin_strategy.CONFIG = qmt_config
|
|
|
|
def test_same_package_binds_same_weights_orders_and_lineage(self):
|
|
package = _package()
|
|
self._install_package(package)
|
|
jq = FakeJoinQuantHarness(
|
|
_jq_histories(),
|
|
positions={
|
|
"600000.XSHG": FakePosition(
|
|
total_amount=1050,
|
|
closeable_amount=200,
|
|
)
|
|
},
|
|
)
|
|
jq.initialize(joinquant_strategy)
|
|
jq_plan = jq.run_scheduled()
|
|
|
|
qmt_position = SimpleNamespace(
|
|
stock_code="600000.SH",
|
|
volume=1050,
|
|
can_use_volume=200,
|
|
)
|
|
context = FakeQmtContext(
|
|
_qmt_histories(),
|
|
positions=[qmt_position],
|
|
bar_time=datetime.datetime(2026, 7, 17, 15, 0),
|
|
)
|
|
qmt = FakeQmtHarness(context)
|
|
qmt_plan = qmt.run(qmt_builtin_strategy)
|
|
|
|
self.assertIsNotNone(jq_plan)
|
|
self.assertIsNotNone(qmt_plan)
|
|
for field in (
|
|
"weights",
|
|
"targets",
|
|
"orders",
|
|
"lineage",
|
|
"universe",
|
|
):
|
|
self.assertEqual(jq_plan[field], qmt_plan[field])
|
|
self.assertEqual(
|
|
jq_plan["lineage"],
|
|
{
|
|
"package_id": package["package_id"],
|
|
"package_sha256": package["package_sha256"],
|
|
"decision_id": package["decision_id"],
|
|
"signal_as_of": package["signal_as_of"],
|
|
"next_session": package["next_session"],
|
|
"release": package["release"],
|
|
"model_id": package["model_id"],
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
jq_plan["execution_contract"],
|
|
{
|
|
"signal_price_source": "CURRENT_DATA_DAY_OPEN",
|
|
"submit_trigger": "DECLARED_NEXT_SESSION_OPEN_CALLBACK",
|
|
"declared_next_session": package["next_session"],
|
|
"real_platform_observed": False,
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
qmt_plan["execution_contract"],
|
|
{
|
|
"signal_price_source": "COMPLETED_DAILY_CLOSE",
|
|
"submit_trigger": "NEXT_BAR_FIRST_TICK",
|
|
"quick_trade": 0,
|
|
"declared_next_session": package["next_session"],
|
|
"real_platform_observed": False,
|
|
},
|
|
)
|
|
self.assertEqual(jq_plan["orders"]["600000.XSHG"], -200)
|
|
self.assertEqual(dict(jq.orders)["600000.XSHG"], 850)
|
|
self.assertEqual(qmt_plan["orders"]["600000.XSHG"], -200)
|
|
sell_order = [
|
|
order
|
|
for order in qmt.orders
|
|
if order["order_code"] == "600000.SH"
|
|
][0]
|
|
self.assertEqual(sell_order["volume"], 200)
|
|
self.assertEqual(sell_order["quick_trade"], 0)
|
|
self.assertTrue(
|
|
all(
|
|
int(quantity) % 100 == 0
|
|
for quantity in jq_plan["targets"].values()
|
|
)
|
|
)
|
|
|
|
def test_missing_exact_clock_is_noop_without_momentum_fallback(self):
|
|
package = _package(
|
|
signal_as_of="2026-07-10T15:00:00+08:00",
|
|
next_session="2026-07-13",
|
|
)
|
|
self._install_package(package)
|
|
jq = FakeJoinQuantHarness(_jq_histories())
|
|
jq.initialize(joinquant_strategy)
|
|
self.assertIsNone(jq.run_scheduled())
|
|
self.assertEqual(jq.orders, [])
|
|
self.assertIsNone(jq.last_index_request)
|
|
self.assertIsNone(jq.last_extras_request)
|
|
self.assertIsNone(jq.last_history_request)
|
|
|
|
context = FakeQmtContext(
|
|
_qmt_histories(),
|
|
bar_time=datetime.datetime(2026, 7, 17, 15, 0),
|
|
)
|
|
qmt = FakeQmtHarness(context)
|
|
self.assertIsNone(qmt.run(qmt_builtin_strategy))
|
|
self.assertEqual(qmt.orders, [])
|
|
self.assertIsNone(context.last_sector_request)
|
|
self.assertIsNone(context.last_market_request)
|
|
|
|
def test_tampered_package_fails_closed_during_initialization(self):
|
|
package = _package()
|
|
tampered = copy.deepcopy(package)
|
|
tampered["target_weights"]["000001.XSHE"] = 0.25
|
|
tampered["constraints"]["actual_gross_weight"] = 0.25
|
|
self._install_package(tampered)
|
|
|
|
jq = FakeJoinQuantHarness(_jq_histories())
|
|
with self.assertRaises(ValueError):
|
|
jq.initialize(joinquant_strategy)
|
|
self.assertEqual(jq.orders, [])
|
|
|
|
context = FakeQmtContext(_qmt_histories())
|
|
qmt = FakeQmtHarness(context)
|
|
with self.assertRaises(ValueError):
|
|
qmt.run(qmt_builtin_strategy)
|
|
self.assertEqual(qmt.orders, [])
|
|
|
|
def test_duplicate_package_id_fails_closed_during_initialization(self):
|
|
first = _package()
|
|
second = _package(
|
|
signal_as_of="2026-07-24T15:00:00+08:00",
|
|
next_session="2026-07-27",
|
|
package_id=first["package_id"],
|
|
decision_id="D-20260724-B",
|
|
)
|
|
self._install_packages([first, second])
|
|
|
|
jq = FakeJoinQuantHarness(_jq_histories())
|
|
with self.assertRaisesRegex(ValueError, "duplicate"):
|
|
jq.initialize(joinquant_strategy)
|
|
|
|
context = FakeQmtContext(_qmt_histories())
|
|
qmt = FakeQmtHarness(context)
|
|
with self.assertRaisesRegex(
|
|
qmt_builtin_strategy.UnsupportedStrategyPeriod,
|
|
"duplicate",
|
|
):
|
|
qmt.run(qmt_builtin_strategy)
|
|
|
|
def test_joinquant_emits_canonical_observation_stream_and_fail_soft_eod(self):
|
|
package = _package()
|
|
self._install_package(package)
|
|
jq = FakeJoinQuantHarness(
|
|
_jq_histories(),
|
|
positions={
|
|
"600000.XSHG": FakePosition(
|
|
total_amount=1050,
|
|
closeable_amount=200,
|
|
)
|
|
},
|
|
)
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
jq.initialize(joinquant_strategy)
|
|
plan = jq.run_scheduled()
|
|
joinquant_strategy.after_trading_end(jq.context)
|
|
self.assertIsNotNone(plan)
|
|
|
|
records = _evidence_records(output)
|
|
events = [record["event"] for record in records]
|
|
for event in (
|
|
"INIT",
|
|
"TARGET_PACKAGE_HIT",
|
|
"PLATFORM_BINDING_INPUT",
|
|
"TARGET_PACKAGE_PLAN",
|
|
"ORDER_REQUEST",
|
|
"ORDER_RETURN",
|
|
"EOD_STATUS",
|
|
):
|
|
self.assertIn(event, events)
|
|
self.assertEqual(
|
|
[record["sequence"] for record in records],
|
|
list(range(1, len(records) + 1)),
|
|
)
|
|
|
|
binding = [
|
|
record["payload"]
|
|
for record in records
|
|
if record["event"] == "PLATFORM_BINDING_INPUT"
|
|
][0]
|
|
self.assertEqual(binding["equity"], 1_000_000.0)
|
|
self.assertEqual(binding["current"], {"600000.XSHG": 1050})
|
|
self.assertEqual(binding["sellable"], {"600000.XSHG": 200})
|
|
self.assertEqual(
|
|
binding["price_source"],
|
|
"CURRENT_DATA_DAY_OPEN",
|
|
)
|
|
self.assertEqual(
|
|
binding["lineage"]["package_sha256"],
|
|
package["package_sha256"],
|
|
)
|
|
|
|
plan_record = [
|
|
record["payload"]
|
|
for record in records
|
|
if record["event"] == "TARGET_PACKAGE_PLAN"
|
|
][0]
|
|
self.assertEqual(plan_record["weights"], package["target_weights"])
|
|
self.assertEqual(plan_record["orders"], plan["orders"])
|
|
|
|
eod_records = [
|
|
record
|
|
for record in records
|
|
if record["event"] == "EOD_STATUS"
|
|
]
|
|
self.assertTrue(eod_records)
|
|
chunk_count = eod_records[0]["chunk_count"]
|
|
self.assertEqual(len(eod_records), chunk_count)
|
|
self.assertEqual(
|
|
[record["chunk_index"] for record in eod_records],
|
|
list(range(1, chunk_count + 1)),
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
record["chunk_count"] == chunk_count
|
|
for record in eod_records
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
record["payload"]["chunk_encoding"]
|
|
== "base64-canonical-json"
|
|
for record in eod_records
|
|
)
|
|
)
|
|
encoded = "".join(
|
|
record["payload"]["content"]
|
|
for record in eod_records
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
record["payload"]["encoded_length"] == len(encoded)
|
|
for record in eod_records
|
|
)
|
|
)
|
|
for record in eod_records:
|
|
line = joinquant_strategy.EVIDENCE_PREFIX + json.dumps(
|
|
record,
|
|
ensure_ascii=True,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
allow_nan=False,
|
|
)
|
|
self.assertLess(len(line.encode("utf-8")), 3 * 1024)
|
|
canonical = base64.b64decode(
|
|
encoded.encode("ascii"),
|
|
validate=True,
|
|
)
|
|
eod = json.loads(canonical.decode("ascii"))
|
|
self.assertEqual(
|
|
base64.b64encode(
|
|
json.dumps(
|
|
eod,
|
|
ensure_ascii=True,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
allow_nan=False,
|
|
).encode("ascii")
|
|
).decode("ascii"),
|
|
encoded,
|
|
)
|
|
self.assertFalse(eod["orders_api"]["available"])
|
|
self.assertFalse(eod["trades_api"]["available"])
|
|
self.assertTrue(eod["portfolio"]["available"])
|
|
self.assertEqual(
|
|
eod["last_plan"]["lineage"]["package_id"],
|
|
package["package_id"],
|
|
)
|
|
|
|
def test_joinquant_target_package_binds_day_open_not_last_price(self):
|
|
package = _package()
|
|
self._install_package(package)
|
|
jq = FakeJoinQuantHarness(_jq_histories())
|
|
jq.current_data["000001.XSHE"].day_open = 20.0
|
|
jq.current_data["000001.XSHE"].last_price = 999.0
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
jq.initialize(joinquant_strategy)
|
|
plan = jq.run_scheduled()
|
|
|
|
self.assertIsNotNone(plan)
|
|
self.assertEqual(plan["targets"]["000001.XSHE"], 15_000)
|
|
self.assertEqual(jq.orders, [("000001.XSHE", 15_000)])
|
|
binding = [
|
|
record["payload"]
|
|
for record in _evidence_records(output)
|
|
if record["event"] == "PLATFORM_BINDING_INPUT"
|
|
][0]
|
|
self.assertEqual(binding["price_source"], "CURRENT_DATA_DAY_OPEN")
|
|
self.assertEqual(binding["prices"]["000001.XSHE"], 20.0)
|
|
self.assertIsNone(jq.last_history_request)
|
|
|
|
def test_joinquant_missing_day_open_fails_closed_without_fallback(self):
|
|
package = _package()
|
|
self._install_package(package)
|
|
jq = FakeJoinQuantHarness(_jq_histories())
|
|
del jq.current_data["000001.XSHE"].day_open
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
jq.initialize(joinquant_strategy)
|
|
with self.assertRaisesRegex(
|
|
joinquant_strategy.PriceHistoryUnavailable,
|
|
"day_open",
|
|
):
|
|
jq.run_scheduled()
|
|
|
|
self.assertEqual(jq.orders, [])
|
|
self.assertIsNone(jq.last_index_request)
|
|
self.assertIsNone(jq.last_extras_request)
|
|
self.assertIsNone(jq.last_history_request)
|
|
events = [
|
|
record["event"] for record in _evidence_records(output)
|
|
]
|
|
self.assertIn("TARGET_PACKAGE_HIT", events)
|
|
self.assertNotIn("PLATFORM_BINDING_INPUT", events)
|
|
self.assertNotIn("TARGET_PACKAGE_PLAN", events)
|
|
self.assertNotIn("ORDER_REQUEST", events)
|
|
|
|
def test_joinquant_noop_eod_clears_prior_session_evidence(self):
|
|
package = _package()
|
|
self._install_package(package)
|
|
jq = FakeJoinQuantHarness(
|
|
_jq_histories(),
|
|
positions={
|
|
"600000.XSHG": FakePosition(
|
|
total_amount=1050,
|
|
closeable_amount=200,
|
|
)
|
|
},
|
|
)
|
|
jq.current_data["000001.XSHE"].paused = True
|
|
|
|
def eod_payload(output):
|
|
chunks = [
|
|
record
|
|
for record in _evidence_records(output)
|
|
if record["event"] == "EOD_STATUS"
|
|
]
|
|
self.assertTrue(chunks)
|
|
self.assertEqual(
|
|
[record["chunk_index"] for record in chunks],
|
|
list(range(1, chunks[0]["chunk_count"] + 1)),
|
|
)
|
|
encoded = "".join(
|
|
record["payload"]["content"] for record in chunks
|
|
)
|
|
return json.loads(
|
|
base64.b64decode(
|
|
encoded.encode("ascii"),
|
|
validate=True,
|
|
).decode("ascii")
|
|
)
|
|
|
|
hit_output = io.StringIO()
|
|
with contextlib.redirect_stdout(hit_output):
|
|
jq.initialize(joinquant_strategy)
|
|
self.assertIsNotNone(jq.run_scheduled())
|
|
joinquant_strategy.after_trading_end(jq.context)
|
|
hit_eod = eod_payload(hit_output)
|
|
self.assertIsNotNone(hit_eod["last_plan"])
|
|
self.assertTrue(hit_eod["order_returns"])
|
|
self.assertTrue(hit_eod["skipped_orders"])
|
|
orders_after_hit = list(jq.orders)
|
|
self.assertTrue(orders_after_hit)
|
|
|
|
for unused_day in range(2):
|
|
del unused_day
|
|
jq.advance_session()
|
|
noop_output = io.StringIO()
|
|
with contextlib.redirect_stdout(noop_output):
|
|
self.assertIsNone(jq.run_scheduled())
|
|
joinquant_strategy.after_trading_end(jq.context)
|
|
noop_records = _evidence_records(noop_output)
|
|
self.assertIn(
|
|
"TARGET_PACKAGE_NOOP",
|
|
[record["event"] for record in noop_records],
|
|
)
|
|
noop_eod = eod_payload(noop_output)
|
|
self.assertIsNone(noop_eod["last_plan"])
|
|
self.assertEqual(noop_eod["order_returns"], [])
|
|
self.assertEqual(noop_eod["skipped_orders"], [])
|
|
self.assertEqual(jq.orders, orders_after_hit)
|
|
|
|
def test_joinquant_hostile_any_still_runs_final_three_day_tape(self):
|
|
bundled = types.ModuleType("hostile_joinquant_bundle")
|
|
source = (
|
|
ROOT / "dist/target-package/joinquant_strategy.py"
|
|
).read_bytes()
|
|
exec(
|
|
compile(
|
|
source,
|
|
"dist/target-package/joinquant_strategy.py",
|
|
"exec",
|
|
),
|
|
bundled.__dict__,
|
|
)
|
|
package = copy.deepcopy(bundled.TARGET_PACKAGES[0])
|
|
histories = {
|
|
symbol: [10.0] * 21 for symbol in package["universe"]
|
|
}
|
|
hostile_any = lambda values: True
|
|
|
|
def run_three_days(module):
|
|
jq = FakeJoinQuantHarness(histories)
|
|
jq.context.current_dt = datetime.datetime(
|
|
2024,
|
|
3,
|
|
11,
|
|
9,
|
|
30,
|
|
)
|
|
jq.context.previous_date = datetime.date(2024, 3, 8)
|
|
daily_orders = []
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
jq.initialize(module)
|
|
for day in range(3):
|
|
before = len(jq.orders)
|
|
jq.run_scheduled()
|
|
daily_orders.append(len(jq.orders) - before)
|
|
if day < 2:
|
|
jq.advance_session()
|
|
self.assertEqual(daily_orders, [0, 4, 0])
|
|
|
|
with self.subTest(artifact="source-wrapper"):
|
|
self._install_package(package)
|
|
with (
|
|
mock.patch.object(
|
|
portable_core,
|
|
"any",
|
|
hostile_any,
|
|
create=True,
|
|
),
|
|
mock.patch.object(
|
|
joinquant_strategy,
|
|
"any",
|
|
hostile_any,
|
|
create=True,
|
|
),
|
|
):
|
|
run_three_days(joinquant_strategy)
|
|
|
|
with self.subTest(artifact="final-target-bundle"):
|
|
bundled.any = hostile_any
|
|
self.assertEqual(
|
|
bundled.TARGET_PACKAGES[0]["package_sha256"],
|
|
package["package_sha256"],
|
|
)
|
|
run_three_days(bundled)
|
|
|
|
def test_joinquant_emits_noop_and_skipped_order(self):
|
|
missing = _package(
|
|
signal_as_of="2026-07-10T15:00:00+08:00",
|
|
next_session="2026-07-13",
|
|
)
|
|
self._install_package(missing)
|
|
jq = FakeJoinQuantHarness(_jq_histories())
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
jq.initialize(joinquant_strategy)
|
|
self.assertIsNone(jq.run_scheduled())
|
|
events = [
|
|
record["event"] for record in _evidence_records(output)
|
|
]
|
|
self.assertIn("TARGET_PACKAGE_NOOP", events)
|
|
self.assertNotIn("TARGET_PACKAGE_PLAN", events)
|
|
self.assertNotIn("ORDER_REQUEST", events)
|
|
|
|
package = _package()
|
|
self._install_package(package)
|
|
jq = FakeJoinQuantHarness(_jq_histories())
|
|
jq.current_data["000001.XSHE"].paused = True
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
jq.initialize(joinquant_strategy)
|
|
plan = jq.run_scheduled()
|
|
self.assertIsNotNone(plan)
|
|
records = _evidence_records(output)
|
|
skipped = [
|
|
record["payload"]
|
|
for record in records
|
|
if record["event"] == "SKIPPED_ORDER"
|
|
]
|
|
self.assertEqual(len(skipped), 1)
|
|
self.assertEqual(skipped[0]["skip"]["reason"], "paused")
|
|
self.assertEqual(jq.orders, [])
|
|
|
|
def test_joinquant_observation_failure_does_not_change_order_path(self):
|
|
package = _package()
|
|
self._install_package(package)
|
|
jq = FakeJoinQuantHarness(_jq_histories())
|
|
with mock.patch("builtins.print", side_effect=RuntimeError("log down")):
|
|
jq.initialize(joinquant_strategy)
|
|
plan = jq.run_scheduled()
|
|
joinquant_strategy.after_trading_end(jq.context)
|
|
self.assertIsNotNone(plan)
|
|
self.assertEqual(
|
|
sorted(jq.orders),
|
|
[("000001.XSHE", 25000)],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|