# coding: utf-8 """JoinQuant hosted wrapper. Upload the generated bundle, not this source file, to JoinQuant. The bundle tool replaces the marked import block with the exact portable core source. """ # __PORTABLE_CORE_BUNDLE_START__ # Inlined by tools/bundle_platforms.py; do not edit this region. import hashlib import json import math import re _CANONICAL_EXCHANGES = { "SH": "XSHG", "SSE": "XSHG", "XSHG": "XSHG", "SZ": "XSHE", "SZE": "XSHE", "SZSE": "XSHE", "XSHE": "XSHE", "BJ": "XBSE", "BSE": "XBSE", "XBSE": "XBSE", } _SHORT_EXCHANGES = { "XSHG": "SH", "XSHE": "SZ", "XBSE": "BJ", } _TARGET_PACKAGE_FIELDS = { "schema_version", "package_type", "package_id", "decision_id", "release", "evidence_class", "model_id", "signal_as_of", "next_session", "model_sha256", "data_sha256", "feature_sha256", "risk_sha256", "cost_sha256", "optimizer_sha256", "config_sha256", "source_sha256", "target_weights", "universe", "constraints", "package_sha256", } _TARGET_PACKAGE_HASH_FIELDS = ( "model_sha256", "data_sha256", "feature_sha256", "risk_sha256", "cost_sha256", "optimizer_sha256", "config_sha256", "source_sha256", ) class TargetPackageLookupMiss(ValueError, LookupError): """An already verified tape has no exact requested decision.""" def _sum_numeric(values): """Sum numerics without trusting a hosted runtime's global ``sum`` name.""" total = 0.0 for value in values: total += value return total def _any_true(values): """Return built-in ``any`` semantics without trusting hosted globals.""" for value in values: if value: return True return False def _infer_exchange(code): if code.startswith(("4", "8")): return "XBSE" if code.startswith(("5", "6", "9")): return "XSHG" return "XSHE" def _split_symbol(symbol): if not isinstance(symbol, str): raise TypeError("symbol must be a string") value = symbol.strip().upper().replace(" ", "") if not value: raise ValueError("symbol must not be empty") match = re.match(r"^(\d{6})\.(XSHG|XSHE|XBSE|SH|SZ|BJ)$", value) if match: return match.group(1), _CANONICAL_EXCHANGES[match.group(2)] match = re.match(r"^(XSHG|XSHE|XBSE|SH|SZ|BJ)[\.:]?(\d{6})$", value) if match: return match.group(2), _CANONICAL_EXCHANGES[match.group(1)] match = re.match(r"^(\d{6})[\.:](SSE|SZE|SZSE|BSE)$", value) if match: return match.group(1), _CANONICAL_EXCHANGES[match.group(2)] if re.match(r"^\d{6}$", value): return value, _infer_exchange(value) raise ValueError("unsupported A-share symbol: %s" % symbol) def normalize_symbol(symbol, target="canonical"): """Convert common A-share codes to canonical/JQ, QMT, or Qlib format.""" code, exchange = _split_symbol(symbol) target_name = str(target).strip().lower() if target_name in ("canonical", "joinquant", "jq"): return "%s.%s" % (code, exchange) if target_name in ("qmt", "xtquant"): return "%s.%s" % (code, _SHORT_EXCHANGES[exchange]) if target_name == "qlib": return "%s%s" % (_SHORT_EXCHANGES[exchange], code) raise ValueError("unsupported target platform: %s" % target) def convert_symbol(symbol, target="canonical"): """Alias kept as the stable adapter-facing name.""" return normalize_symbol(symbol, target) def _canonical_json(value): """Match ``quant60.ledger.canonical_json`` for plain JSON values.""" return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ) def _json_copy(value, name): try: return json.loads(_canonical_json(value)) except (TypeError, ValueError) as error: raise ValueError("%s must contain finite JSON values only" % name) from error def _canonical_mapping(value, name, value_validator): if not hasattr(value, "items"): raise TypeError("%s must be a mapping" % name) output = {} for raw_symbol, raw_value in value.items(): symbol = normalize_symbol(raw_symbol) if symbol in output: raise ValueError( "%s contains duplicate normalized symbol %s" % (name, symbol) ) output[symbol] = value_validator(raw_value, "%s.%s" % (name, symbol)) return dict((symbol, output[symbol]) for symbol in sorted(output)) def _finite_non_negative(value, name): if isinstance(value, bool): raise ValueError("%s must be a finite non-negative number" % name) try: numeric = float(value) except (TypeError, ValueError): raise ValueError("%s must be a finite non-negative number" % name) if not math.isfinite(numeric) or numeric < 0.0: raise ValueError("%s must be a finite non-negative number" % name) return numeric def _non_negative_quantity(value, name): numeric = _finite_non_negative(value, name) if numeric != math.floor(numeric): raise ValueError("%s must be a whole-share quantity" % name) return int(numeric) def _positive_price(value, name): numeric = _finite_non_negative(value, name) if numeric <= 0.0: raise ValueError("%s must be positive" % name) return numeric def _optional_non_empty_string(value, name): if value is None: return None if not isinstance(value, str) or not value.strip(): raise ValueError("%s must be a non-empty string when provided" % name) return value def _sha256_hex(value, name, allow_none=False): if value is None and allow_none: return None if ( not isinstance(value, str) or len(value) != 64 or _any_true( character not in "0123456789abcdef" for character in value ) ): raise ValueError( "%s must be a 64-character lowercase SHA-256 digest" % name ) if value == "0" * 64: raise ValueError("%s cannot be an all-zero placeholder" % name) return value def _reject_quantity_fields(value, path="$"): if hasattr(value, "items"): for raw_key, child in value.items(): key = str(raw_key) normalized = key.lower().replace("-", "_") if "quantity" in normalized or "quantities" in normalized: raise ValueError( "%s.%s: quantities do not belong in TargetPackageV1" % (path, key) ) _reject_quantity_fields(child, "%s.%s" % (path, key)) elif isinstance(value, list): for index, child in enumerate(value): _reject_quantity_fields(child, "%s[%d]" % (path, index)) def _verify_identifier(value, name): if ( not isinstance(value, str) or re.match(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$", value) is None ): raise ValueError("%s is not a canonical identifier" % name) return value def _verify_target_package_clock(signal_as_of, next_session): if ( not isinstance(signal_as_of, str) or re.match( r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T15:00:00\+08:00$", signal_as_of, ) is None ): raise ValueError("signal_as_of must be canonical China-market close") if ( not isinstance(next_session, str) or re.match(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}$", next_session) is None ): raise ValueError("next_session must use canonical YYYY-MM-DD") if next_session <= signal_as_of[:10]: raise ValueError("next_session must follow signal_as_of") def _target_package_content_sha256(package): body = dict(package) body.pop("package_sha256", None) _reject_quantity_fields(body) try: payload = _canonical_json(body).encode("utf-8") except (TypeError, ValueError) as error: raise ValueError( "target package must contain finite JSON values only" ) from error return hashlib.sha256(payload).hexdigest() def verify_target_package( package, expected_signal_as_of=None, expected_next_session=None, expected_decision_id=None, expected_package_sha256=None, ): """Verify the platform-critical subset of a sealed ``TargetPackageV1``. This intentionally duplicates a small, Python-3.6-compatible safety boundary from the richer local validator. Hosted runtimes must verify the embedded package again instead of trusting the bundling process. """ if not hasattr(package, "items"): raise TypeError("target package must be a mapping") candidate = _json_copy(package, "target package") _reject_quantity_fields(candidate) fields = set(candidate) if fields != _TARGET_PACKAGE_FIELDS: missing = sorted(_TARGET_PACKAGE_FIELDS - fields) unknown = sorted(fields - _TARGET_PACKAGE_FIELDS) raise ValueError( "target package fields mismatch; missing=%s, unknown=%s" % (missing, unknown) ) if candidate["schema_version"] != "1.0": raise ValueError("unsupported target package schema_version") if candidate["package_type"] != "TargetPackageV1": raise ValueError("unsupported target package type") for name in ( "package_id", "decision_id", "release", "evidence_class", "model_id", ): _verify_identifier(candidate[name], name) _verify_target_package_clock( candidate["signal_as_of"], candidate["next_session"] ) for name in _TARGET_PACKAGE_HASH_FIELDS: _sha256_hex(candidate[name], name) _sha256_hex(candidate["package_sha256"], "package_sha256") raw_weights = candidate["target_weights"] if not hasattr(raw_weights, "items") or not raw_weights: raise ValueError("target_weights must be a non-empty mapping") weights = {} for raw_symbol, raw_weight in raw_weights.items(): symbol = normalize_symbol(raw_symbol) if symbol != raw_symbol: raise ValueError( "target_weights symbol must be canonical: %s" % raw_symbol ) if symbol in weights: raise ValueError("duplicate target weight for %s" % symbol) weight = _finite_non_negative( raw_weight, "target_weights.%s" % symbol ) if weight > 1.0: raise ValueError("target weight exceeds one for %s" % symbol) weights[symbol] = weight gross = math.fsum(weights.values()) if gross > 1.0 + 1e-12: raise ValueError("target_weights gross exceeds one") universe = candidate["universe"] if not hasattr(universe, "items") or not universe: raise ValueError("universe must be a non-empty mapping") canonical_universe = {} for raw_symbol, raw_state in universe.items(): symbol = normalize_symbol(raw_symbol) if symbol != raw_symbol: raise ValueError("universe symbol must be canonical: %s" % raw_symbol) if symbol in canonical_universe: raise ValueError("duplicate universe symbol %s" % symbol) if not hasattr(raw_state, "items"): raise ValueError("universe state must be a mapping for %s" % symbol) state = raw_state.get("state") if state not in ( "openable", "hold_only", "sell_only", "frozen", "excluded", ): raise ValueError("unsupported universe state for %s" % symbol) canonical_universe[symbol] = state missing_universe = sorted(set(weights) - set(canonical_universe)) if missing_universe: raise ValueError( "target symbols are absent from universe: %s" % missing_universe ) for symbol, weight in weights.items(): if canonical_universe[symbol] == "excluded" and weight > 1e-12: raise ValueError("excluded symbol has positive weight: %s" % symbol) constraints = candidate["constraints"] if not hasattr(constraints, "items"): raise ValueError("constraints must be a mapping") if constraints.get("long_only") is not True: raise ValueError("TargetPackageV1 must be long-only") declared_gross = _finite_non_negative( constraints.get("actual_gross_weight"), "constraints.actual_gross_weight", ) max_gross = _finite_non_negative( constraints.get("max_gross_weight"), "constraints.max_gross_weight", ) max_single = _finite_non_negative( constraints.get("max_single_weight"), "constraints.max_single_weight", ) if abs(declared_gross - gross) > 1e-12: raise ValueError("declared gross does not equal target_weights gross") if gross > max_gross + 1e-12 or max_gross > 1.0: raise ValueError("target_weights exceed max_gross_weight") if max_single <= 0.0 or max_single > 1.0: raise ValueError("invalid max_single_weight") if _any_true( weight > max_single + 1e-12 for weight in weights.values() ): raise ValueError("target_weights exceed max_single_weight") model_claim = " ".join( ( candidate["release"], candidate["evidence_class"], ) ).lower() if ( "baseline" in model_claim and "portable-momentum" in candidate["model_id"].lower() ): raise ValueError( "portable-momentum cannot claim a baseline target package" ) expected_hash = _target_package_content_sha256(candidate) if candidate["package_sha256"] != expected_hash: raise ValueError("target package content hash mismatch") if ( expected_signal_as_of is not None and candidate["signal_as_of"] != expected_signal_as_of ): raise ValueError("target package signal_as_of does not match request") if ( expected_next_session is not None and candidate["next_session"] != expected_next_session ): raise ValueError("target package next_session does not match request") if ( expected_decision_id is not None and candidate["decision_id"] != expected_decision_id ): raise ValueError("target package decision_id does not match request") if expected_package_sha256 is not None: _sha256_hex( expected_package_sha256, "expected_package_sha256", ) if candidate["package_sha256"] != expected_package_sha256: raise ValueError("target package digest does not match request") return candidate def target_package_tape_sha256(packages): """Hash the exact ordered canonical JSON package array.""" if not isinstance(packages, (list, tuple)): raise TypeError("target package tape must be a list or tuple") try: payload = _canonical_json(list(packages)).encode("utf-8") except (TypeError, ValueError) as error: raise ValueError( "target package tape must contain finite JSON values only" ) from error return hashlib.sha256(payload).hexdigest() def lookup_target_package_exact( packages, signal_as_of, next_session, decision_id=None, expected_tape_sha256=None, expected_count=None, ): """Verify a complete tape, then return one exact dual-clock decision.""" if not isinstance(packages, (list, tuple)) or not packages: raise ValueError("target package tape must be a non-empty list or tuple") _verify_target_package_clock(signal_as_of, next_session) decision_id = _optional_non_empty_string(decision_id, "decision_id") if expected_count is not None: if ( isinstance(expected_count, bool) or not isinstance(expected_count, int) or expected_count < 0 ): raise ValueError("expected_count must be a non-negative integer") if len(packages) != expected_count: raise ValueError("target package tape count mismatch") if expected_tape_sha256 is not None: _sha256_hex(expected_tape_sha256, "expected_tape_sha256") if target_package_tape_sha256(packages) != expected_tape_sha256: raise ValueError("target package tape digest mismatch") verified = [] package_ids = set() decision_ids = set() clock_keys = set() for raw_package in packages: package = verify_target_package(raw_package) clock_key = (package["signal_as_of"], package["next_session"]) if package["package_id"] in package_ids: raise ValueError("duplicate target package_id") if package["decision_id"] in decision_ids: raise ValueError("duplicate target decision_id") if clock_key in clock_keys: raise ValueError("duplicate target package clock") package_ids.add(package["package_id"]) decision_ids.add(package["decision_id"]) clock_keys.add(clock_key) verified.append(package) matches = [ package for package in verified if package["signal_as_of"] == signal_as_of and package["next_session"] == next_session ] if not matches: raise TargetPackageLookupMiss( "no target package for exact signal_as_of/next_session clock" ) if len(matches) != 1: raise ValueError("ambiguous exact target package decision") if ( decision_id is not None and matches[0]["decision_id"] != decision_id ): raise ValueError("exact target package decision_id mismatch") return _json_copy(matches[0], "target package") def momentum_score(prices, lookback=20, skip=0): """Return trailing simple momentum, or ``None`` if history is insufficient. ``skip=0`` uses the latest completed decision bar. Set ``skip=1`` only when intentionally implementing a 20-1 convention. A lookback of N needs N+skip+1 observations. """ lookback = int(lookback) skip = int(skip) if lookback <= 0 or skip < 0: raise ValueError("lookback must be positive and skip non-negative") values = list(prices) if len(values) < lookback + skip + 1: return None end_index = len(values) - 1 - skip start_index = end_index - lookback start = float(values[start_index]) end = float(values[end_index]) if ( start <= 0.0 or end <= 0.0 or not math.isfinite(start) or not math.isfinite(end) ): return None return end / start - 1.0 def rank_momentum(price_history, lookback=20, skip=0, top_n=None): """Rank ``{symbol: closes}`` by momentum with deterministic tie-breaking.""" ranked = [] for symbol in sorted(price_history): score = momentum_score(price_history[symbol], lookback, skip) if score is not None and math.isfinite(score): ranked.append((normalize_symbol(symbol), score)) ranked.sort(key=lambda item: (-item[1], item[0])) if top_n is not None: ranked = ranked[: max(0, int(top_n))] return ranked def capped_target_weights( scores, max_weight=0.20, gross_target=1.0, min_score=0.0 ): """Long-only score-proportional weights with iterative cap redistribution.""" max_weight = float(max_weight) gross_target = float(gross_target) min_score = float(min_score) if not math.isfinite(max_weight) or not 0.0 < max_weight <= 1.0: raise ValueError("max_weight must be finite and lie in (0, 1]") if not math.isfinite(gross_target) or not 0.0 <= gross_target <= 1.0: raise ValueError("gross_target must be finite and lie in [0, 1]") if not math.isfinite(min_score): raise ValueError("min_score must be finite") if hasattr(scores, "items"): items = list(scores.items()) else: items = list(scores) normalized = {} for symbol, raw_score in items: name = normalize_symbol(symbol) score = float(raw_score) if not math.isfinite(score): raise ValueError("score must be finite for %s" % name) normalized[name] = score weights = dict((symbol, 0.0) for symbol in normalized) active = dict( (symbol, score) for symbol, score in normalized.items() if score > min_score and score > 0.0 ) remaining = min(gross_target, len(active) * max_weight) while active and remaining > 1e-15: score_sum = _sum_numeric(active.values()) if score_sum <= 0.0: break capped = [] for symbol in sorted(active): proposed = remaining * active[symbol] / score_sum if proposed >= max_weight - 1e-15: weights[symbol] = max_weight capped.append(symbol) if not capped: for symbol in sorted(active): weights[symbol] = remaining * active[symbol] / score_sum remaining = 0.0 break for symbol in capped: active.pop(symbol) remaining -= max_weight remaining = max(0.0, remaining) return weights def round_board_lot(quantity, lot_size=100): """Round an absolute long position down to a board-lot quantity.""" lot_size = int(lot_size) if lot_size <= 0: raise ValueError("lot_size must be positive") if not math.isfinite(float(quantity)): raise ValueError("quantity must be finite") quantity = max(0, int(quantity)) return quantity // lot_size * lot_size def _is_star_market(symbol): code, exchange = _split_symbol(symbol) return exchange == "XSHG" and code.startswith(("688", "689")) def target_quantities( weights, prices, equity, lot_size=100, cash_buffer=0.02 ): """Convert target weights to affordable board-lot long quantities.""" equity = float(equity) cash_buffer = float(cash_buffer) if ( not math.isfinite(equity) or equity < 0.0 or not math.isfinite(cash_buffer) or not 0.0 <= cash_buffer < 1.0 ): raise ValueError("invalid equity or cash_buffer") positive_weights = [] for raw_symbol, raw_weight in weights.items(): weight = float(raw_weight) if not math.isfinite(weight): raise ValueError( "weight must be finite for %s" % normalize_symbol(raw_symbol) ) positive_weights.append(max(0.0, weight)) requested_gross = _sum_numeric(positive_weights) gross_ceiling = 1.0 - cash_buffer scale = ( min(1.0, gross_ceiling / requested_gross) if requested_gross > 0.0 else 1.0 ) result = {} for raw_symbol in sorted(weights): symbol = normalize_symbol(raw_symbol) weight = float(weights[raw_symbol]) if not math.isfinite(weight): raise ValueError("weight must be finite for %s" % symbol) weight = max(0.0, weight) price = prices.get(raw_symbol) if price is None: price = prices.get(symbol) if price is None: raise KeyError("missing price for %s" % symbol) price = float(price) if price <= 0.0 or not math.isfinite(price): raise ValueError("invalid price for %s" % symbol) # Weights are absolute equity weights. cash_buffer is a ceiling, not # a second multiplicative haircut: gross=.95 and buffer=.02 stays .95. budget = equity * weight * scale quantity = round_board_lot(budget / price, lot_size) # STAR Market auction orders must contain at least 200 shares. Keep # the portable planner conservative and on the configured 100-share # grid even though quantities above 200 may increment by one share. if _is_star_market(symbol) and 0 < quantity < 200: quantity = 0 result[symbol] = quantity return result def order_deltas( targets, current, sellable=None, lot_size=100 ): """Return signed board-lot orders; negative sells respect sellable quantity.""" canonical_targets = {} canonical_current = {} canonical_sellable = {} for symbol, quantity in targets.items(): canonical_targets[normalize_symbol(symbol)] = max(0, int(quantity)) for symbol, quantity in current.items(): canonical_current[normalize_symbol(symbol)] = max(0, int(quantity)) if sellable is not None: for symbol, quantity in sellable.items(): canonical_sellable[normalize_symbol(symbol)] = max(0, int(quantity)) result = {} symbols = sorted(set(canonical_targets) | set(canonical_current)) for symbol in symbols: target = round_board_lot(canonical_targets.get(symbol, 0), lot_size) if _is_star_market(symbol) and 0 < target < 200: target = 0 held = canonical_current.get(symbol, 0) delta = target - held if delta > 0: delta = round_board_lot(delta, lot_size) if _is_star_market(symbol) and 0 < delta < 200: delta = 0 elif delta < 0: available = held if sellable is not None: available = min(held, canonical_sellable.get(symbol, 0)) requested = min(-delta, available) # A complete liquidation may submit the entire odd-lot balance. # This also covers the STAR exception for a remaining balance # below 200 shares. Partial orders stay on the configured grid. full_liquidation = ( target == 0 and requested == held and available >= held ) if full_liquidation: delta = -held else: delta = -round_board_lot(requested, lot_size) if _is_star_market(symbol) and 0 < -delta < 200: delta = 0 if delta: result[symbol] = delta return result def build_target_weight_plan( target_weights, prices, current, sellable, equity, cash_buffer=0.02, lot_size=100, decision_id=None, package_sha256=None, ): """Bind canonical post-risk weights to one account without recomputing Alpha. ``prices`` must cover exactly every target or currently managed symbol. Missing prices fail closed even for a held symbol whose desired weight is zero. A missing ``sellable`` entry is safe and means zero sellable shares. """ weights = _canonical_mapping( target_weights, "target_weights", _finite_non_negative, ) gross = math.fsum(weights.values()) if _any_true(weight > 1.0 for weight in weights.values()): raise ValueError("target weight must not exceed one") if gross > 1.0 + 1e-12: raise ValueError("target weight gross must not exceed one") canonical_current = _canonical_mapping( current, "current", _non_negative_quantity, ) canonical_sellable = _canonical_mapping( sellable, "sellable", _non_negative_quantity, ) unknown_sellable = sorted(set(canonical_sellable) - set(canonical_current)) if unknown_sellable: raise ValueError( "sellable contains symbols absent from current: %s" % unknown_sellable ) for symbol, quantity in canonical_sellable.items(): if quantity > canonical_current[symbol]: raise ValueError( "sellable exceeds current quantity for %s" % symbol ) canonical_prices = _canonical_mapping( prices, "prices", _positive_price, ) managed_symbols = set(weights) | set(canonical_current) price_symbols = set(canonical_prices) missing_prices = sorted(managed_symbols - price_symbols) extra_prices = sorted(price_symbols - managed_symbols) if missing_prices or extra_prices: if missing_prices: raise KeyError( "prices do not cover managed symbols: missing=%s" % missing_prices ) raise ValueError( "prices contain unmanaged symbols: extra=%s" % extra_prices ) complete_weights = dict( (symbol, weights.get(symbol, 0.0)) for symbol in sorted(managed_symbols) ) targets = target_quantities( complete_weights, canonical_prices, equity, lot_size, cash_buffer, ) orders = order_deltas( targets, canonical_current, canonical_sellable, lot_size, ) decision_id = _optional_non_empty_string(decision_id, "decision_id") package_sha256 = _sha256_hex( package_sha256, "package_sha256", allow_none=True, ) return { "decision_id": decision_id, "package_sha256": package_sha256, "weights": complete_weights, "targets": targets, "orders": orders, } def build_rebalance_plan( price_history, current, sellable, equity, lookback=20, skip=0, top_n=10, max_weight=0.20, gross_target=1.0, cash_buffer=0.02, lot_size=100, ): """Build the complete portable signal -> weight -> quantity -> order plan.""" ranked = rank_momentum(price_history, lookback, skip, top_n) scores = dict(ranked) weights = capped_target_weights(scores, max_weight, gross_target, 0.0) for raw_symbol in price_history: symbol = normalize_symbol(raw_symbol) if symbol not in scores: scores[symbol] = momentum_score( price_history[raw_symbol], lookback, skip ) if symbol not in weights: weights[symbol] = 0.0 prices = {} for raw_symbol, history in price_history.items(): if not history: continue prices[normalize_symbol(raw_symbol)] = history[-1] targets = target_quantities( weights, prices, equity, lot_size, cash_buffer ) orders = order_deltas(targets, current, sellable, lot_size) return { "scores": scores, "weights": weights, "targets": targets, "orders": orders, } __all__ = [ "build_rebalance_plan", "build_target_weight_plan", "capped_target_weights", "convert_symbol", "lookup_target_package_exact", "momentum_score", "normalize_symbol", "order_deltas", "rank_momentum", "round_board_lot", "TargetPackageLookupMiss", "target_package_tape_sha256", "target_quantities", "verify_target_package", ] # __PORTABLE_CORE_BUNDLE_END__ import base64 as _q60_base64 import json as _q60_json import math class PriceHistoryUnavailable(RuntimeError): pass class UnmanagedPositionError(RuntimeError): pass class ClockStateError(RuntimeError): pass # __BASELINE_CONFIG_START__ # Generated from configs/baseline.json; do not edit this region. CONFIG = {'decision_mode': 'target_package', 'target_package_count': 1, 'target_package_tape_sha256': '5383c376951637b8d4b1ca42fc6db80347abda74c0181c0f70d6c411fb97ea64', 'universe_mode': 'fixed', 'index_symbol': '000905.XSHG', 'qmt_sector_name': '\u4e2d\u8bc1500', 'dedicated_account_required': True, 'exclude_st': True, 'universe': ['000001.XSHE', '000333.XSHE', '000651.XSHE', '002415.XSHE', '002594.XSHE', '300059.XSHE', '300750.XSHE', '600000.XSHG', '600519.XSHG', '601012.XSHG', '601318.XSHG', '688981.XSHG'], 'lookback': 20, 'skip': 0, 'top_n': 20, 'max_weight': 0.05, 'gross_target': 0.95, 'cash_buffer': 0.02, 'lot_size': 100, 'participation_rate': 0.1, 'slippage_bps': 2.0, 'commission_rate': 0.0002, 'minimum_commission': 5.0, 'stamp_duty_rate': 0.0005, 'transfer_fee_rate': 1e-05, 'rebalance_schedule': 'weekly_first_close'} # __BASELINE_CONFIG_END__ # __TARGET_PACKAGE_TAPE_START__ # Generated by tools/bundle_platforms.py; do not edit this region. TARGET_PACKAGES = [{'config_sha256': 'e4edf1eb730597e220266c1f8007ddb5e1ab4d26a490cd10738823527d030470', 'constraints': {'actual_gross_weight': 0.4, 'constraint_state': 'FEASIBLE', 'long_only': True, 'max_gross_weight': 0.9, 'max_one_way_turnover': 0.45, 'max_single_weight': 0.1}, 'cost_sha256': 'e2a2d8e6d3829c105a9f6c9ec80d9df18c2dee0553aafeb761abb79d378769a7', 'data_sha256': 'f5b3e66b0a94eeccb44a114bf3a0bbfad44b5d5da9d5b698c81e83a3e0007d03', 'decision_id': 'RD-20240311', 'evidence_class': 'synthetic-research', 'feature_sha256': '24e671cebc35afe7497b69c59d48bb5bad9401e30c1607f0fc41a69f59082aa1', 'model_id': 'deterministic-ridge-alpha-5', 'model_sha256': 'e14e8df7e2f86e6b3178a6e500d1d2c0427bb02d187b13cb4ec672bbf5f9f334', 'next_session': '2024-03-12', 'optimizer_sha256': 'b6aba6ab22fc375e89e4bfbdffb10c7261a2fae01b0183265caacb589643c31e', 'package_id': 'TP-20240311-795dcfba', 'package_sha256': 'baa15c8156aa6cb54d3775d0cd562613a9d0a4f15b0c39f2a60e0f8b8c6f1f3a', 'package_type': 'TargetPackageV1', 'release': 'quant60-research-candidate-v1', 'risk_sha256': 'a3d95aa6cd569d2452856b148df053b040b2acc2e6b993bbd1072b48b46cde5a', 'schema_version': '1.0', 'signal_as_of': '2024-03-11T15:00:00+08:00', 'source_sha256': '86c1e0514f608b2012b4084dbc947eada90e8015fccbdf8ae13ae99340ede178', 'target_weights': {'000001.XSHE': 0.0, '000333.XSHE': 0.0, '000651.XSHE': 0.0, '002415.XSHE': 0.0, '002594.XSHE': 0.0, '300059.XSHE': 0.0, '300750.XSHE': 0.0, '600000.XSHG': 0.0, '600519.XSHG': 0.1, '601012.XSHG': 0.1, '601318.XSHG': 0.1, '688981.XSHG': 0.1}, 'universe': {'000001.XSHE': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '000333.XSHE': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '000651.XSHE': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '002415.XSHE': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '002594.XSHE': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '300059.XSHE': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '300750.XSHE': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '600000.XSHG': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '600519.XSHG': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '601012.XSHG': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '601318.XSHG': {'reasons': ['ELIGIBLE'], 'state': 'openable'}, '688981.XSHG': {'reasons': ['ELIGIBLE'], 'state': 'openable'}}}] # __TARGET_PACKAGE_TAPE_END__ MOMENTUM_SMOKE_MODE = "portable_momentum_smoke" TARGET_PACKAGE_MODE = "target_package" EVIDENCE_PREFIX = "QUANT60_JOINQUANT_EVIDENCE_V1 " EVIDENCE_CHUNK_CHARACTERS = 1800 TARGET_PRICE_SOURCE = "CURRENT_DATA_DAY_OPEN" def _evidence_value(value, depth=0): """Return stable JSON data without allowing observation to stop trading.""" if depth > 8: return "" if value is None or isinstance(value, (bool, int, str)): return value if isinstance(value, float): return value if math.isfinite(value) else None if isinstance(value, dict): return { str(key): _evidence_value(item, depth + 1) for key, item in sorted( value.items(), key=lambda pair: str(pair[0]), ) } if isinstance(value, (list, tuple)): return [_evidence_value(item, depth + 1) for item in value] isoformat = getattr(value, "isoformat", None) if callable(isoformat): try: return str(isoformat()) except Exception: return "" try: return str(value) except Exception: return "" def _evidence_context_time(context): try: return _evidence_value(getattr(context, "current_dt", None)) except Exception: return None def _emit_evidence( event, payload=None, context=None, chunk_index=None, chunk_count=None, ): """Print one prefixed canonical JSON record and always fail soft.""" try: sequence = int( getattr(g, "quant60_evidence_sequence", 0) or 0 ) + 1 g.quant60_evidence_sequence = sequence record = { "context_time": _evidence_context_time(context), "decision_mode": getattr( g, "quant60_decision_mode", CONFIG.get("decision_mode"), ), "event": str(event), "payload": _evidence_value(payload or {}), "schema_version": "1.0", "sequence": sequence, "source": "joinquant_hosted", } if chunk_index is not None or chunk_count is not None: if ( type(chunk_index) is not int or type(chunk_count) is not int or chunk_index <= 0 or chunk_count <= 0 or chunk_index > chunk_count ): return None record["chunk_index"] = chunk_index record["chunk_count"] = chunk_count encoded = _q60_json.dumps( record, ensure_ascii=True, sort_keys=True, separators=(",", ":"), allow_nan=False, ) print(EVIDENCE_PREFIX + encoded) except Exception: # Evidence is observational. A logging/serialization failure must # never alter an already verified decision or hosted order path. return None return record def _emit_chunked_evidence(event, payload, context=None): """Emit reconstructable base64 chunks of one canonical JSON payload.""" try: canonical = _q60_json.dumps( _evidence_value(payload), ensure_ascii=True, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode("ascii") encoded = _q60_base64.b64encode(canonical).decode("ascii") chunks = [ encoded[index : index + EVIDENCE_CHUNK_CHARACTERS] for index in range(0, len(encoded), EVIDENCE_CHUNK_CHARACTERS) ] or [""] count = len(chunks) for index, content in enumerate(chunks, 1): _emit_evidence( event, { "chunk_encoding": "base64-canonical-json", "content": content, "encoded_length": len(encoded), }, context, chunk_index=index, chunk_count=count, ) except Exception: return None return count def _observe_object(value, fields): if value is None: return None output = {"object_type": type(value).__name__} for name in fields: try: item = getattr(value, name) except Exception: continue if callable(item): continue output[name] = _evidence_value(item) return output def _observe_order(value): return _observe_object( value, ( "order_id", "add_time", "security", "amount", "filled", "price", "avg_cost", "side", "action", "status", "target_quantity", ), ) def _observe_trade(value): return _observe_object( value, ( "trade_id", "order_id", "time", "security", "amount", "price", ), ) def _observe_platform_collection(api_name, observer): """Read an optional hosted collection without assuming API availability.""" function = globals().get(api_name) if not callable(function): return { "available": False, "reason": "API_UNAVAILABLE", } try: raw = function() except Exception as exc: return { "available": False, "error_type": type(exc).__name__, "reason": "API_CALL_FAILED", } if hasattr(raw, "items"): items = [ { "key": str(key), "value": observer(value), } for key, value in sorted( raw.items(), key=lambda pair: str(pair[0]), ) ] elif isinstance(raw, (list, tuple)): items = [observer(value) for value in raw] elif raw is None: items = [] else: items = [observer(raw)] return { "available": True, "count": len(items), "items": items, } def _observe_portfolio(context): try: portfolio = context.portfolio except Exception as exc: return { "available": False, "error_type": type(exc).__name__, } output = { "available": True, "fields": {}, "positions": [], } for name in ( "total_value", "cash", "available_cash", "positions_value", "returns", ): try: value = getattr(portfolio, name) except Exception: continue output["fields"][name] = _evidence_value(value) try: positions = getattr(portfolio, "positions", {}) if hasattr(positions, "items"): for symbol, position in sorted( positions.items(), key=lambda pair: str(pair[0]), ): item = _observe_object( position, ( "total_amount", "closeable_amount", "price", "avg_cost", "value", ), ) item["symbol"] = str(symbol) output["positions"].append(item) except Exception as exc: output["positions_error_type"] = type(exc).__name__ return output def _plan_evidence(plan): if not isinstance(plan, dict): return None return { "decision_mode": plan.get("decision_mode"), "execution_contract": plan.get("execution_contract"), "lineage": plan.get("lineage"), "orders": plan.get("orders"), "targets": plan.get("targets"), "universe": plan.get("universe"), "weights": plan.get("weights"), } def _validated_target_packages(config): mode = config.get("decision_mode") if mode not in (MOMENTUM_SMOKE_MODE, TARGET_PACKAGE_MODE): raise ValueError("unsupported decision_mode") if not isinstance(TARGET_PACKAGES, list): raise ValueError("TARGET_PACKAGES must be a list") expected_count = config.get("target_package_count") expected_sha256 = config.get("target_package_tape_sha256") if mode == MOMENTUM_SMOKE_MODE: if ( type(expected_count) is not int or expected_count != 0 or expected_sha256 is not None or TARGET_PACKAGES ): raise ValueError( "portable_momentum_smoke cannot carry target packages" ) return [] if type(expected_count) is not int or expected_count <= 0: raise ValueError( "target_package mode requires a positive target_package_count" ) if expected_count != len(TARGET_PACKAGES): raise ValueError("target package count mismatch") actual_sha256 = target_package_tape_sha256(TARGET_PACKAGES) if expected_sha256 != actual_sha256: raise ValueError("target package tape sha256 mismatch") verified = [] package_ids = set() decision_ids = set() clocks = set() for raw_package in TARGET_PACKAGES: package = verify_target_package(raw_package) package_id = package["package_id"] decision_id = package["decision_id"] clock = (package["signal_as_of"], package["next_session"]) if ( package_id in package_ids or decision_id in decision_ids or clock in clocks ): raise ValueError("duplicate target package decision identity") package_ids.add(package_id) decision_ids.add(decision_id) clocks.add(clock) verified.append(package) return verified def initialize(context): """Configure the selected daily-open decision mode.""" set_option("avoid_future_data", True) set_option("use_real_price", True) g.quant60_evidence_sequence = 0 g.quant60_order_observations = [] g.quant60_config = dict(CONFIG) g.quant60_target_packages = _validated_target_packages( g.quant60_config ) g.quant60_decision_mode = g.quant60_config["decision_mode"] if g.quant60_config["decision_mode"] == MOMENTUM_SMOKE_MODE: if g.quant60_config.get("rebalance_schedule") != "weekly_first_close": raise ValueError("unsupported rebalance_schedule") if g.quant60_config.get("universe_mode") not in ("pit_index", "fixed"): raise ValueError("unsupported universe_mode") if ( g.quant60_config.get("universe_mode") == "pit_index" and not g.quant60_config.get("dedicated_account_required") ): raise ValueError("pit_index mode requires a dedicated account") set_benchmark(g.quant60_config["index_symbol"]) commission = ( float(g.quant60_config["commission_rate"]) + float(g.quant60_config["transfer_fee_rate"]) ) set_order_cost( OrderCost( open_tax=0.0, close_tax=float(g.quant60_config["stamp_duty_rate"]), open_commission=commission, close_commission=commission, close_today_commission=0.0, min_commission=float( g.quant60_config["minimum_commission"] ), ), type="stock", ) # JoinQuant applies half of PriceRelatedSlippage on each side. The local # slippage_bps setting is one-sided, hence the explicit factor of two. set_slippage( PriceRelatedSlippage( 2.0 * float(g.quant60_config["slippage_bps"]) / 10000.0 ), type="stock", ) set_option( "order_volume_ratio", float(g.quant60_config["participation_rate"]), ) g.quant60_last_plan = None g.quant60_last_universe = None g.quant60_last_universe_as_of = None g.quant60_skipped_orders = [] g.quant60_pending_first_session = None g.quant60_last_callback_date = None run_daily(rebalance, time="open") _emit_evidence( "INIT", { "benchmark": g.quant60_config["index_symbol"], "package_count": len(g.quant60_target_packages), "package_ids": [ package["package_id"] for package in g.quant60_target_packages ], "rebalance_callback": "run_daily_open", "target_package_tape_sha256": g.quant60_config.get( "target_package_tape_sha256" ), }, context, ) def _decision_mode(): configured = g.quant60_config.get("decision_mode") frozen = getattr(g, "quant60_decision_mode", configured) if configured != frozen: raise ValueError("decision_mode cannot change after initialize") return frozen def _close_history(symbol, count): raw = attribute_history( symbol, count, unit="1d", fields=["close"], skip_paused=False, df=False, fq="pre", ) values = list(raw.get("close", [])) if hasattr(raw, "get") else [] if len(values) != int(count): raise PriceHistoryUnavailable( "%s needs %d completed daily closes, got %d" % (symbol, int(count), len(values)) ) output = [] for value in values: try: number = float(value) except (TypeError, ValueError): raise PriceHistoryUnavailable( "%s contains a missing/non-numeric close" % symbol ) if not math.isfinite(number) or number <= 0.0: raise PriceHistoryUnavailable( "%s contains a missing/non-positive close" % symbol ) output.append(number) return output def _close_histories(symbols, count): raw = history( int(count), unit="1d", field="close", security_list=list(symbols), df=False, skip_paused=False, fq="pre", ) if not hasattr(raw, "get"): raise PriceHistoryUnavailable("history returned an invalid payload") output = {} for symbol in symbols: values = list(raw.get(symbol, [])) if len(values) != int(count): raise PriceHistoryUnavailable( "%s needs %d completed daily closes, got %d" % (symbol, int(count), len(values)) ) clean = [] for value in values: try: number = float(value) except (TypeError, ValueError): raise PriceHistoryUnavailable( "%s contains a missing/non-numeric close" % symbol ) if not math.isfinite(number) or number <= 0.0: raise PriceHistoryUnavailable( "%s contains a missing/non-positive close" % symbol ) clean.append(number) output[convert_symbol(symbol, "canonical")] = clean return output def _decision_universe(context): config = g.quant60_config if config.get("universe_mode") == "fixed": values = list(config["universe"]) else: as_of = _as_date( getattr(context, "previous_date", None), "previous_date", ) values = get_index_stocks( config["index_symbol"], date=as_of, ) if not values: raise PriceHistoryUnavailable( "no PIT index members for %s on %s" % (config["index_symbol"], as_of.isoformat()) ) g.quant60_last_universe_as_of = as_of.isoformat() normalized = sorted( set(convert_symbol(symbol, "joinquant") for symbol in values) ) if not normalized: raise PriceHistoryUnavailable("decision universe is empty") g.quant60_last_universe = list(normalized) return normalized def _filter_st_universe(context, symbols): if not g.quant60_config.get("exclude_st", True): return list(symbols) as_of = _as_date( getattr(context, "previous_date", None), "previous_date", ) raw = get_extras( "is_st", list(symbols), end_date=as_of, count=1, df=False, ) if not hasattr(raw, "get"): raise PriceHistoryUnavailable( "get_extras is_st returned an invalid payload" ) output = [] for symbol in symbols: values = list(raw.get(symbol, [])) if len(values) != 1: raise PriceHistoryUnavailable( "%s needs one PIT is_st value on %s" % (symbol, as_of.isoformat()) ) try: numeric = float(values[0]) except (TypeError, ValueError): raise PriceHistoryUnavailable( "%s contains a non-boolean PIT is_st value" % symbol ) if not math.isfinite(numeric) or numeric not in (0.0, 1.0): raise PriceHistoryUnavailable( "%s contains an invalid PIT is_st value" % symbol ) if not bool(numeric): output.append(symbol) if not output: raise PriceHistoryUnavailable( "PIT ST exclusion removed the entire decision universe" ) g.quant60_last_universe = list(output) return output def _position_amount(position, name, default=0): value = getattr(position, name, default) return int(value or 0) def _snapshot_portfolio(context, managed): current = {} sellable = {} managed = set(convert_symbol(symbol, "canonical") for symbol in managed) dynamic = ( _decision_mode() == MOMENTUM_SMOKE_MODE and g.quant60_config.get("universe_mode") == "pit_index" ) positions = getattr(context.portfolio, "positions", {}) for symbol, position in positions.items(): canonical = convert_symbol(str(symbol), "canonical") quantity = _position_amount(position, "total_amount") if quantity and canonical not in managed and not dynamic: raise UnmanagedPositionError( "account contains unmanaged position %s; use a dedicated " "strategy account or reconcile ownership before trading" % canonical ) if canonical not in managed and not dynamic: continue current[canonical] = quantity sellable[canonical] = _position_amount( position, "closeable_amount", current[canonical] ) equity = float(getattr(context.portfolio, "total_value", 0.0) or 0.0) return current, sellable, equity def _is_star_board(symbol): code = str(symbol).split(".", 1)[0] return code.startswith("688") or code.startswith("689") def _order_request(symbol, delta): """Build a fail-closed hosted order request from current market data. JoinQuant rejects an unprotected market order for STAR Market securities. A limit at the exchange daily bound supplies equivalent worst-price protection while keeping the rebalance target expressed in exact shares. """ try: item = get_current_data()[symbol] except Exception as exc: g.quant60_skipped_orders.append( {"symbol": symbol, "reason": "current_data_unavailable", "detail": str(exc)} ) return None if bool(getattr(item, "paused", False)): g.quant60_skipped_orders.append( {"symbol": symbol, "reason": "paused"} ) return None if not _is_star_board(symbol): return {"style": None} field = "high_limit" if int(delta) > 0 else "low_limit" raw_price = getattr(item, field, None) try: protection_price = float(raw_price) except (TypeError, ValueError): protection_price = float("nan") if ( not math.isfinite(protection_price) or protection_price <= 0.0 or protection_price >= 10000.0 ): g.quant60_skipped_orders.append( { "symbol": symbol, "reason": "star_protection_price_invalid", "detail": field, } ) return None try: style = LimitOrderStyle(protection_price) except Exception as exc: g.quant60_skipped_orders.append( { "symbol": symbol, "reason": "star_protection_style_unavailable", "detail": str(exc), } ) return None return {"style": style} def _target_package_for_open(context): current_date = _as_date(getattr(context, "current_dt", None), "current_dt") previous_date = _as_date( getattr(context, "previous_date", None), "previous_date", ) signal_as_of = ( previous_date.isoformat() + "T15:00:00+08:00" ) next_session = current_date.isoformat() config = g.quant60_config packages = _validated_target_packages(config) matches = [ package for package in packages if package["signal_as_of"] == signal_as_of and package["next_session"] == next_session ] if not matches: _emit_evidence( "TARGET_PACKAGE_NOOP", { "fallback_invoked": False, "next_session": next_session, "orders_submitted": 0, "package_count": len(packages), "reason": "NO_EXACT_CLOCK_MATCH", "signal_as_of": signal_as_of, "target_package_tape_sha256": config[ "target_package_tape_sha256" ], }, context, ) return None if len(matches) != 1: raise ValueError("ambiguous target package clock") package = lookup_target_package_exact( packages, signal_as_of, next_session, decision_id=matches[0]["decision_id"], expected_tape_sha256=config["target_package_tape_sha256"], expected_count=config["target_package_count"], ) _emit_evidence( "TARGET_PACKAGE_HIT", { "lineage": _target_lineage(package), "target_package_tape_sha256": config[ "target_package_tape_sha256" ], }, context, ) return package def _target_prices(symbols): try: current_data = get_current_data() except Exception as exc: raise PriceHistoryUnavailable( "current_data is unavailable: %s" % exc ) prices = {} for canonical in sorted(symbols): symbol = convert_symbol(canonical, "joinquant") try: raw_price = getattr(current_data[symbol], "day_open") price = float(raw_price) except (KeyError, AttributeError, TypeError, ValueError): raise PriceHistoryUnavailable( "%s has no usable current day_open" % symbol ) if not math.isfinite(price) or price <= 0.0: raise PriceHistoryUnavailable( "%s has no usable current day_open" % symbol ) prices[canonical] = price return prices def _target_lineage(package): return { "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"], } def _compute_target_package_plan(context): package = _target_package_for_open(context) if package is None: return None managed = sorted(package["universe"]) current, sellable, equity = _snapshot_portfolio(context, managed) if equity <= 0: raise ValueError("portfolio.total_value must be positive") prices = _target_prices( set(package["target_weights"]) | set(current) ) _emit_evidence( "PLATFORM_BINDING_INPUT", { "current": current, "equity": equity, "lineage": _target_lineage(package), "price_source": TARGET_PRICE_SOURCE, "prices": prices, "sellable": sellable, }, context, ) plan = build_target_weight_plan( target_weights=package["target_weights"], prices=prices, current=current, sellable=sellable, equity=equity, cash_buffer=g.quant60_config["cash_buffer"], lot_size=g.quant60_config["lot_size"], decision_id=package["decision_id"], package_sha256=package["package_sha256"], ) plan["decision_mode"] = TARGET_PACKAGE_MODE plan["lineage"] = _target_lineage(package) plan["universe"] = package["universe"] plan["execution_contract"] = { "signal_price_source": TARGET_PRICE_SOURCE, "submit_trigger": "DECLARED_NEXT_SESSION_OPEN_CALLBACK", "declared_next_session": package["next_session"], "real_platform_observed": False, } _emit_evidence( "TARGET_PACKAGE_PLAN", _plan_evidence(plan), context, ) return plan def compute_plan(context): """Return the selected portable plan without submitting orders.""" if _decision_mode() == TARGET_PACKAGE_MODE: return _compute_target_package_plan(context) config = g.quant60_config required = int(config["lookback"]) + int(config["skip"]) + 1 platform_universe = _decision_universe(context) platform_universe = _filter_st_universe( context, platform_universe, ) price_history = _close_histories(platform_universe, required) current, sellable, equity = _snapshot_portfolio( context, platform_universe, ) if equity <= 0: raise ValueError("portfolio.total_value must be positive") plan = build_rebalance_plan( price_history=price_history, current=current, sellable=sellable, equity=equity, lookback=config["lookback"], skip=config["skip"], top_n=config["top_n"], max_weight=config["max_weight"], gross_target=config["gross_target"], cash_buffer=config["cash_buffer"], lot_size=config["lot_size"], ) return plan def _submit_plan(context, plan, managed): """Submit a portable plan's exact executable share deltas.""" if plan is None: return None observe_target_package = ( plan.get("decision_mode") == TARGET_PACKAGE_MODE ) order_deltas = plan["orders"] current, _, unused_equity = _snapshot_portfolio( context, managed, ) del unused_equity all_symbols = set(current) all_symbols.update(order_deltas) # Exits first. The requested target is current + executable delta, not the # unconstrained portfolio target; this preserves T+1 sellable caps and odd # lots from the portable plan. for canonical in sorted( all_symbols, key=lambda item: ( int(order_deltas.get(item, 0)) > 0, item, ), ): delta = int(order_deltas.get(canonical, 0)) if delta == 0: continue platform_symbol = convert_symbol(canonical, "joinquant") skipped_before = len(g.quant60_skipped_orders) request = _order_request(platform_symbol, delta) if request is None: skipped = g.quant60_skipped_orders[skipped_before:] if not skipped: skipped = [ { "symbol": platform_symbol, "reason": "ORDER_REQUEST_NOT_BUILT", } ] if observe_target_package: for item in skipped: _emit_evidence( "SKIPPED_ORDER", { "canonical_symbol": canonical, "delta": delta, "lineage": plan.get("lineage"), "skip": item, }, context, ) continue target = int(current.get(canonical, 0)) + delta request_evidence = { "canonical_symbol": canonical, "current_quantity": int(current.get(canonical, 0)), "delta": delta, "lineage": plan.get("lineage"), "platform_symbol": platform_symbol, "style": _observe_object( request["style"], ("kind", "limit_price"), ), "target_quantity": target, } if observe_target_package: _emit_evidence( "ORDER_REQUEST", request_evidence, context, ) try: if request["style"] is None: order_result = order_target(platform_symbol, target) else: order_result = order_target( platform_symbol, target, style=request["style"], ) except Exception as exc: if observe_target_package: _emit_evidence( "ORDER_RETURN", { "error_type": type(exc).__name__, "lineage": plan.get("lineage"), "outcome": "EXCEPTION", "platform_symbol": platform_symbol, "target_quantity": target, }, context, ) raise observation = { "lineage": plan.get("lineage"), "order": _observe_order(order_result), "outcome": "RETURNED", "platform_symbol": platform_symbol, "target_quantity": target, } if observe_target_package: g.quant60_order_observations.append(observation) _emit_evidence( "ORDER_RETURN", observation, context, ) g.quant60_last_plan = plan return plan def _execute_rebalance(context): """Submit the selected portable plan's exact executable share deltas.""" plan = compute_plan(context) if plan is None: return None managed = ( sorted(plan["universe"]) if "universe" in plan else ( g.quant60_last_universe or g.quant60_config["universe"] ) ) return _submit_plan(context, plan, managed) def _as_date(value, name): if hasattr(value, "date"): value = value.date() if not hasattr(value, "isocalendar"): raise ClockStateError("%s must be a date/datetime" % name) return value def rebalance(context): """Execute on the session immediately after a week's first close. A daily state machine is used instead of ``run_weekly(..., 2)`` so a one-session holiday week still executes on the next available session. """ current_date = _as_date(getattr(context, "current_dt", None), "current_dt") previous_date = _as_date( getattr(context, "previous_date", None), "previous_date", ) if g.quant60_last_callback_date == current_date: return None if _decision_mode() == TARGET_PACKAGE_MODE: # EOD evidence is session-scoped. Clear the prior session only after # the duplicate-date guard so a same-day retry cannot erase evidence. g.quant60_last_plan = None g.quant60_order_observations = [] g.quant60_skipped_orders = [] # Consume the daily clock token before crossing the hosted order # boundary so callback retries cannot duplicate an accepted prefix. g.quant60_last_callback_date = current_date return _execute_rebalance(context) pending = g.quant60_pending_first_session is_first_session = ( previous_date.isocalendar()[:2] != current_date.isocalendar()[:2] ) if pending is not None: if previous_date != pending: raise ClockStateError( "pending first-session close is not the immediately " "previous trading session" ) # Consume the clock token before crossing any hosted order boundary. # A partial API failure must not make a callback retry duplicate the # already accepted prefix of the order list. g.quant60_pending_first_session = ( current_date if is_first_session else None ) g.quant60_last_callback_date = current_date plan = _execute_rebalance(context) return plan if is_first_session: g.quant60_pending_first_session = current_date g.quant60_last_callback_date = current_date return None def after_trading_end(context): """Emit a best-effort end-of-day observation without changing state.""" if _decision_mode() != TARGET_PACKAGE_MODE: return None return _emit_chunked_evidence( "EOD_STATUS", { "last_plan": _plan_evidence( getattr(g, "quant60_last_plan", None) ), "order_returns": getattr( g, "quant60_order_observations", [], ), "orders_api": _observe_platform_collection( "get_orders", _observe_order, ), "portfolio": _observe_portfolio(context), "skipped_orders": getattr( g, "quant60_skipped_orders", [], ), "trades_api": _observe_platform_collection( "get_trades", _observe_trade, ), }, context, )