diff --git a/backend_api_python/app/commands/strategy_dispatcher_worker.py b/backend_api_python/app/commands/strategy_dispatcher_worker.py
index afa8c7bba..b9b5a8c36 100644
--- a/backend_api_python/app/commands/strategy_dispatcher_worker.py
+++ b/backend_api_python/app/commands/strategy_dispatcher_worker.py
@@ -10,6 +10,7 @@ def main() -> None:
os.environ["QD_PROCESS_ROLE"] = "strategy-dispatcher"
from app import create_app
+ from app.events.bar_recovery import StrategyBarRecoveryPublisher
from app.events.dispatcher import StrategyDueDispatcher
from app.events.kafka import KafkaEventConsumer
from app.events.protocol import MarketBarClosedV1
@@ -26,6 +27,7 @@ def main() -> None:
heartbeat_repository = StrategyCommandRepository()
worker_id = build_worker_id()
dispatcher = StrategyDueDispatcher()
+ recovery = StrategyBarRecoveryPublisher()
consumer = KafkaEventConsumer(
topics=[topic_for_event(MarketBarClosedV1.event_type)],
group_id=os.getenv(
@@ -34,23 +36,46 @@ def main() -> None:
),
)
last_heartbeat = 0.0
+ last_recovery = 0.0
+ try:
+ recovery_interval = max(
+ 1.0,
+ float(os.getenv("STRATEGY_BAR_RECOVERY_INTERVAL_SEC", "5")),
+ )
+ except (TypeError, ValueError):
+ recovery_interval = 5.0
+
+ def dispatch(event):
+ recovery.observe(event)
+ return dispatcher.dispatch(event)
with app.app_context():
try:
logger.info("Strategy dispatcher worker started: %s", worker_id)
while not shutdown.event.is_set():
- consumer.poll_once(dispatcher.dispatch, timeout=1.0)
+ consumer.poll_once(dispatch, timeout=1.0)
now = time.monotonic()
+ if now - last_recovery >= recovery_interval:
+ try:
+ recovery.publish_due()
+ except Exception:
+ logger.exception("Strategy bar recovery cycle failed")
+ last_recovery = now
if now - last_heartbeat >= 10.0:
heartbeat_repository.record_worker_heartbeat(
worker_id=worker_id,
role="strategy-dispatcher",
- metadata={**consumer.snapshot(), **dispatcher.snapshot()},
+ metadata={
+ **consumer.snapshot(),
+ **dispatcher.snapshot(),
+ **recovery.snapshot(),
+ },
)
last_heartbeat = now
finally:
consumer.close()
dispatcher.close()
+ recovery.close()
heartbeat_repository.mark_worker_stopped(worker_id)
logger.info("Strategy dispatcher worker stopped: %s", worker_id)
diff --git a/backend_api_python/app/data_sources/asia_stock_kline.py b/backend_api_python/app/data_sources/asia_stock_kline.py
index 23c330b33..3e52ea1eb 100644
--- a/backend_api_python/app/data_sources/asia_stock_kline.py
+++ b/backend_api_python/app/data_sources/asia_stock_kline.py
@@ -145,7 +145,7 @@ def _get_twelve_data_api_key() -> str:
"15m": "15min",
"30m": "30min",
"1H": "1h",
- "4H": "4h",
+ "4H": "1h",
"1D": "1day",
"1W": "1week",
}
@@ -231,6 +231,8 @@ def fetch_twelvedata_klines(
symbol, exchange = _td_symbol_and_exchange(tencent_code, is_hk)
merge_factor = _MERGE_FACTOR_MAP.get(timeframe, 1)
+ if is_hk and timeframe in {"1H", "4H"}:
+ interval, merge_factor = "30min", 2 if timeframe == "1H" else 8
params: Dict[str, Any] = {
"symbol": symbol,
"exchange": exchange,
@@ -242,6 +244,8 @@ def fetch_twelvedata_klines(
}
if before_time:
end_dt = datetime.fromtimestamp(int(before_time))
+ if is_hk:
+ end_dt = pd.Timestamp(int(before_time), unit="s", tz="UTC").tz_convert("Asia/Hong_Kong")
params["end_date"] = end_dt.strftime("%Y-%m-%d %H:%M:%S")
url = "https://api.twelvedata.com/time_series"
@@ -290,7 +294,10 @@ def fetch_twelvedata_klines(
dt_str = v.get("datetime", "")
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
try:
- ts = int(datetime.strptime(dt_str, fmt).timestamp())
+ parsed_time = datetime.strptime(dt_str, fmt)
+ if is_hk:
+ parsed_time = pd.Timestamp(parsed_time).tz_localize("Asia/Hong_Kong")
+ ts = int(parsed_time.timestamp())
break
except ValueError:
continue
@@ -316,7 +323,7 @@ def fetch_twelvedata_klines(
out.sort(key=lambda x: x["time"])
if merge_factor > 1 and out:
- out = _merge_every_n_sorted_bars(out, merge_factor)
+ out = _merge_session_bars(out, timeframe, is_hk, source_seconds=1800 if is_hk and timeframe in {"1H", "4H"} else None)
logger.debug("TwelveData returned %d bars for %s/%s tf=%s", len(out), symbol, exchange, timeframe)
return out
@@ -483,7 +490,7 @@ def fetch_yfinance_klines(
bars = _bars_from_yfinance_df(df)
if merge_factor > 1 and bars:
- bars = _merge_every_n_sorted_bars(bars, merge_factor)
+ bars = _merge_session_bars(bars, timeframe, is_hk)
logger.debug("yfinance returned %d bars for %s tf=%s", len(bars), yf_sym, timeframe)
return bars
@@ -513,7 +520,7 @@ def _min_bar_window(
return start.strftime(fmt), end.strftime(fmt)
-def _bars_from_ak_min_df(df: Any) -> List[Dict[str, Any]]:
+def _bars_from_ak_min_df(df: Any, *, is_hk: bool = False, period_seconds: int = 0) -> List[Dict[str, Any]]:
if df is None or getattr(df, "empty", True):
return []
cols = [str(x) for x in df.columns]
@@ -536,6 +543,9 @@ def _pick(name_zh: str, idx: int) -> str:
for _, row in df.iterrows():
try:
t = pd.Timestamp(row[time_c])
+ if is_hk:
+ t = t.tz_localize("Asia/Hong_Kong") if t.tzinfo is None else t
+ t -= pd.Timedelta(seconds=period_seconds)
ts = int(t.timestamp())
o, c, h, low, v = float(row[c_open]), float(row[c_close]), float(row[c_high]), float(row[c_low]), float(row[c_vol])
out.append({
@@ -552,6 +562,15 @@ def _pick(name_zh: str, idx: int) -> str:
return out
+def _merge_session_bars(bars, timeframe, is_hk, *, source_seconds=None):
+ if not is_hk:
+ return _merge_every_n_sorted_bars(bars, _MERGE_FACTOR_MAP.get(timeframe, 1))
+ from app.services.market_schedule import resample_equity_klines
+ source_seconds = source_seconds or (3600 if timeframe == "4H" else 60)
+ target_seconds = {"1H": 3600, "4H": 14400, "3m": 180}[timeframe]
+ return resample_equity_klines(bars, "HKStock", source_seconds, target_seconds)
+
+
def _merge_every_n_sorted_bars(bars: List[Dict[str, Any]], n: int) -> List[Dict[str, Any]]:
if n <= 1 or len(bars) < n:
return bars
@@ -587,6 +606,8 @@ def fetch_akshare_minute_klines(
return []
p = _minute_period_str(timeframe)
+ if is_hk and timeframe in {"1H", "4H"}:
+ p = "30"
if p is None:
return []
try:
@@ -597,6 +618,11 @@ def fetch_akshare_minute_klines(
sym = ak_hk_code_from_tencent(tencent_code) if is_hk else ak_a_code_from_tencent(tencent_code)
sd, ed = _min_bar_window(timeframe, limit, before_time, after_time)
+ if is_hk:
+ if before_time is not None:
+ ed = pd.Timestamp(before_time, unit="s", tz="UTC").tz_convert("Asia/Hong_Kong").strftime("%Y-%m-%d %H:%M:%S")
+ if after_time is not None:
+ sd = pd.Timestamp(after_time, unit="s", tz="UTC").tz_convert("Asia/Hong_Kong").strftime("%Y-%m-%d %H:%M:%S")
adj = "" if p == "1" else "qfq"
df: Any = None
@@ -625,10 +651,12 @@ def fetch_akshare_minute_klines(
logger.warning("AkShare minute K-line failed %s tf=%s sym=%s: %s", tencent_code, timeframe, sym, e)
return []
- bars = _bars_from_ak_min_df(df)
+ bars = _bars_from_ak_min_df(df, is_hk=is_hk, period_seconds=int(p) * 60)
merge_factor = _MERGE_FACTOR_MAP.get(timeframe, 1)
+ if is_hk and timeframe in {"1H", "4H"}:
+ merge_factor = 2 if timeframe == "1H" else 8
if merge_factor > 1 and bars:
- bars = _merge_every_n_sorted_bars(bars, merge_factor)
+ bars = _merge_session_bars(bars, timeframe, is_hk, source_seconds=int(p) * 60)
return bars
diff --git a/backend_api_python/app/data_sources/candle_calendar.py b/backend_api_python/app/data_sources/candle_calendar.py
new file mode 100644
index 000000000..ca6ca19db
--- /dev/null
+++ b/backend_api_python/app/data_sources/candle_calendar.py
@@ -0,0 +1,24 @@
+"""Native crypto candle boundaries, independent of strategy and transport code."""
+
+from app.data_sources.base import timeframe_seconds
+
+
+def candle_origin(timeframe: str, venue: str = "", market_type: str = "spot") -> int:
+ seconds = timeframe_seconds(timeframe)
+ venue = str(venue or "").lower()
+ offset = -8 * 3600 if venue in {"okx", "bitget", "htx", "huobi"} else 0
+ if seconds == 604800:
+ if venue == "gate" and market_type == "swap":
+ return 0
+ offset += 4 * 86400
+ return offset % seconds
+
+
+def candle_open(timestamp: float, timeframe: str, venue: str = "", market_type: str = "spot") -> int:
+ seconds = timeframe_seconds(timeframe)
+ origin = candle_origin(timeframe, venue, market_type)
+ return int((timestamp - origin) // seconds) * seconds + origin
+
+
+def completed_candle_open(timestamp: float, timeframe: str, venue: str = "", market_type: str = "spot") -> int:
+ return candle_open(timestamp, timeframe, venue, market_type) - timeframe_seconds(timeframe)
diff --git a/backend_api_python/app/data_sources/crypto.py b/backend_api_python/app/data_sources/crypto.py
index 19899e221..5ca01d8d3 100644
--- a/backend_api_python/app/data_sources/crypto.py
+++ b/backend_api_python/app/data_sources/crypto.py
@@ -501,11 +501,14 @@ def get_kline(
limit: int,
before_time: Optional[int] = None,
after_time: Optional[int] = None,
+ *,
+ closed_only: bool = False,
) -> List[Dict[str, Any]]:
"""获取加密货币K线数据"""
self._clear_last_failure()
klines = []
symbol_pair = ""
+ candle_options = {"closed_only": True} if closed_only else {}
# A public source is cached and reused by chart/backtest/signal
# runtimes. Once its configured venue has failed and another venue has
@@ -530,7 +533,7 @@ def get_kline(
preferred_rows = type(self).for_exchange(
preferred_exchange,
fallback_market_type,
- ).get_kline(symbol, timeframe, limit, before_time, after_time)
+ ).get_kline(symbol, timeframe, limit, before_time, after_time, **candle_options)
if preferred_rows:
return preferred_rows
# The promoted provider has also become unavailable. Clear it and
@@ -591,6 +594,7 @@ def get_kline(
ohlcv = self._fetch_ohlcv(
symbol_pair, fetch_exchange_timeframe, fetch_limit,
before_time, fetch_qd_timeframe, after_time,
+ **candle_options,
)
if not ohlcv:
@@ -728,6 +732,7 @@ def get_kline(
limit,
before_time,
after_time,
+ **candle_options,
)
if fallback_rows:
self._preferred_public_exchange_id = exchange_id
@@ -825,6 +830,8 @@ def _fetch_ohlcv(
before_time: Optional[int],
timeframe: str,
after_time: Optional[int] = None,
+ *,
+ closed_only: bool = False,
) -> List:
"""获取OHLCV数据（支持分页获取完整数据）"""
try:
@@ -949,6 +956,7 @@ def _fetch_ohlcv(
since=current_since,
limit=request_limit,
until=end_ms,
+ **({"closed_only": True} if closed_only else {}),
)
else:
batch = self.exchange.fetch_ohlcv(
@@ -1000,7 +1008,8 @@ def _fetch_ohlcv(
ohlcv = sorted(by_ts.values(), key=lambda r: r[0])
if not ohlcv:
return self._fetch_ohlcv_fallback(
- symbol_pair, exchange_timeframe, limit, before_time, timeframe, after_time
+ symbol_pair, exchange_timeframe, limit, before_time, timeframe, after_time,
+ **({"closed_only": True} if closed_only else {}),
)
# Never attach an old partial page across an unreported gap. Only
# validate the returned chart tail, not its extra warmup buffer.
@@ -1046,7 +1055,10 @@ def _fetch_ohlcv(
# rejecting the request outright or silently truncating, which
# downstream code then interprets as "empty data".
safe_limit = min(int(limit), self._SINGLE_FETCH_HARD_CAP)
- ohlcv = self.exchange.fetch_ohlcv(symbol_pair, exchange_timeframe, limit=safe_limit)
+ ohlcv = self.exchange.fetch_ohlcv(
+ symbol_pair, exchange_timeframe, limit=safe_limit,
+ **({"closed_only": True} if closed_only and isinstance(self.exchange, NativeCryptoPublicClient) else {}),
+ )
return ohlcv
@@ -1077,7 +1089,8 @@ def _fetch_ohlcv(
logger.warning(f"Exchange fetch_ohlcv failed: {str(e)}; trying fallback")
self._set_last_failure(e, symbol=symbol_pair, timeframe=timeframe)
return self._fetch_ohlcv_fallback(
- symbol_pair, exchange_timeframe, limit, before_time, timeframe, after_time
+ symbol_pair, exchange_timeframe, limit, before_time, timeframe, after_time,
+ **({"closed_only": True} if closed_only else {}),
)
def _fetch_ohlcv_fallback(
@@ -1088,6 +1101,8 @@ def _fetch_ohlcv_fallback(
before_time: Optional[int],
timeframe: str,
after_time: Optional[int] = None,
+ *,
+ closed_only: bool = False,
) -> List:
"""备用获取方法"""
# An explicit historical window must never be replaced with a recent
@@ -1121,7 +1136,10 @@ def _fetch_ohlcv_fallback(
# exactly when it mattered. Cap it so we at least return one valid
# page of data, which downstream callers can then handle gracefully.
safe_limit = min(int(limit), self._SINGLE_FETCH_HARD_CAP)
- ohlcv = self.exchange.fetch_ohlcv(symbol_pair, exchange_timeframe, since=since, limit=safe_limit)
+ ohlcv = self.exchange.fetch_ohlcv(
+ symbol_pair, exchange_timeframe, since=since, limit=safe_limit,
+ **({"closed_only": True} if closed_only and isinstance(self.exchange, NativeCryptoPublicClient) else {}),
+ )
if ohlcv:
return ohlcv
except Exception as e:
@@ -1137,6 +1155,7 @@ def _fetch_ohlcv_fallback(
symbol_pair,
exchange_timeframe,
limit=min(int(limit), self._SINGLE_FETCH_HARD_CAP),
+ **({"closed_only": True} if closed_only and isinstance(self.exchange, NativeCryptoPublicClient) else {}),
)
if recent:
logger.warning(
diff --git a/backend_api_python/app/data_sources/factory.py b/backend_api_python/app/data_sources/factory.py
index 936f7ead4..9b59bca71 100644
--- a/backend_api_python/app/data_sources/factory.py
+++ b/backend_api_python/app/data_sources/factory.py
@@ -268,10 +268,16 @@ def get_kline_with_diagnostics(
try:
assert_fd_available(f"market-data kline {m}:{symbol}")
source = cls._resolve_source(m, exchange_id=exchange_id, market_type=market_type)
- klines = source.get_kline(symbol, timeframe, limit, before_time, after_time)
+ from app.data_sources.crypto import CryptoDataSource
+ candle_options = {"closed_only": True} if closed_only and isinstance(source, CryptoDataSource) else {}
+ klines = source.get_kline(symbol, timeframe, limit, before_time, after_time, **candle_options)
if closed_only:
- klines = filter_closed_klines(klines, timeframe)
+ if m in {"USStock", "HKStock"}:
+ from app.services.market_schedule import filter_closed_equity_klines
+ klines = filter_closed_equity_klines(klines, timeframe, m)
+ else:
+ klines = filter_closed_klines(klines, timeframe)
klines.sort(key=lambda x: x['time'])
failure = None
diff --git a/backend_api_python/app/data_sources/native_crypto.py b/backend_api_python/app/data_sources/native_crypto.py
index a6c9ec3b0..eae79c04b 100644
--- a/backend_api_python/app/data_sources/native_crypto.py
+++ b/backend_api_python/app/data_sources/native_crypto.py
@@ -8,6 +8,8 @@
import requests
+from app.data_sources.candle_calendar import candle_open
+
SUPPORTED_EXCHANGES = ("binance", "bitget", "bybit", "okx", "gate", "htx")
@@ -529,6 +531,7 @@ def fetch_ohlcv(
since: Optional[int] = None,
limit: int = 300,
until: Optional[int] = None,
+ closed_only: bool = False,
) -> List[List[Any]]:
if timeframe not in self.timeframes:
raise NativeCryptoAPIError(f"Unsupported timeframe {timeframe} on {self.id}")
@@ -613,7 +616,7 @@ def fetch_ohlcv(
# Bitget returns candles before endTime. Move to the next
# interval boundary so the requested final candle is included.
params["endTime"] = min(
- (page_end // tf_ms + 1) * tf_ms,
+ candle_open(page_end / 1000, timeframe, self.id, self.market_type) * 1000 + tf_ms,
int(since) + max_window_ms,
now_ms,
)
@@ -625,7 +628,13 @@ def fetch_ohlcv(
params = {"interval": interval, "limit": min(limit, 1000)}
if since is not None:
params["from"] = int(since // 1000)
- params["to"] = page_end // 1000
+ # Futures aggregates trades only through `to`, not the whole
+ # candle whose opening timestamp equals `to`.
+ params["to"] = min(
+ candle_open(page_end / 1000, timeframe, self.id, self.market_type)
+ + tf_ms // 1000 - 1,
+ upper_bound // 1000,
+ ) if self.market_type == "swap" else page_end // 1000
params.pop("limit", None)
if self.market_type == "swap":
params["contract"] = native
@@ -671,6 +680,13 @@ def fetch_ohlcv(
else:
raise ValueError(f"Unsupported crypto exchange: {self.id}")
lower_bound = int(since) if since is not None else 0
+ if closed_only:
+ unconfirmed = set()
+ if self.id == "okx":
+ unconfirmed = {int(row[0]) for row in rows if len(row) > 8 and str(row[8]) != "1"}
+ elif self.id == "gate" and self.market_type == "spot":
+ unconfirmed = {int(row[0]) * 1000 for row in rows if len(row) > 7 and str(row[7]).lower() != "true"}
+ parsed = [row for row in parsed if row[0] not in unconfirmed and row[0] + tf_ms <= upper_bound]
unique = {
int(row[0]): row for row in parsed
if row and int(row[0]) >= lower_bound
diff --git a/backend_api_python/app/data_sources/us_stock.py b/backend_api_python/app/data_sources/us_stock.py
index 1993021dc..9ffe79bc0 100644
--- a/backend_api_python/app/data_sources/us_stock.py
+++ b/backend_api_python/app/data_sources/us_stock.py
@@ -515,9 +515,13 @@ def get_kline(
if after_time is None and before_time is None:
klines = self._fetch_nasdaq_intraday_chart(symbol, timeframe, effective_limit)
else:
- klines = self._fetch_nasdaq_historical(symbol, start_date, end_date, effective_limit)
+ klines = self._fetch_nasdaq_historical(
+ symbol, start_date, end_date,
+ effective_limit * 5 if timeframe == '1W' else effective_limit,
+ )
if timeframe == '1W' and klines:
- klines = self._merge_every_n_sorted_bars(klines, 5)
+ from app.services.market_schedule import resample_equity_klines
+ klines = resample_equity_klines(klines, "USStock", 86400, 604800)
df = None
if not klines:
df = self._fetch_yfinance(symbol, interval, start_date, end_date)
diff --git a/backend_api_python/app/events/bar_clock.py b/backend_api_python/app/events/bar_clock.py
index 688d83492..8cfdc6789 100644
--- a/backend_api_python/app/events/bar_clock.py
+++ b/backend_api_python/app/events/bar_clock.py
@@ -10,8 +10,8 @@
from app.events.bus import EventBus, EventSubscription
from app.events.protocol import EventEnvelope, MarketBarClosedV1
-from app.services.market_schedule import is_equity_bar_close
-from app.services.strategy_runtime.timeframes import completed_bar_token, seconds_until_next_completed_bar
+from app.services.market_schedule import equity_data_market, is_equity_bar_close
+from app.services.strategy_runtime.timeframes import completed_bar_token, completed_bar_close, seconds_until_next_completed_bar
from app.services.strategy_v2.frequencies import frequency_seconds, normalize_frequency
from app.utils.logger import get_logger
@@ -36,7 +36,12 @@ def from_member(
*,
default_venue: str = "",
) -> "BarStreamKey":
- market = str(member.get("underlying_market") or member.get("market") or "unknown").strip()
+ source_market = str(member.get("market") or "unknown").strip()
+ market = equity_data_market(
+ source_market,
+ member,
+ str(member.get("api_family") or ""),
+ )
venue = str(
member.get("exchange_id")
or member.get("data_source")
@@ -131,7 +136,7 @@ def subscribe(
)
with self._lock:
self._ref_counts[key] = self._ref_counts.get(key, 0) + 1
- self._last_tokens.setdefault(key, completed_bar_token(key.timeframe, self._now()))
+ self._last_tokens.setdefault(key, completed_bar_token(key.timeframe, self._now(), venue=key.venue, market_type=key.market_type))
if self._autostart:
self._start_locked()
self._wake.set()
@@ -159,7 +164,7 @@ def publish_due(self, now: datetime | None = None) -> int:
published = 0
session_boundaries: dict[tuple[str, datetime], bool] = {}
for key in keys:
- token = completed_bar_token(key.timeframe, current)
+ token = completed_bar_token(key.timeframe, current, venue=key.venue, market_type=key.market_type)
with self._lock:
previous = self._last_tokens.get(key)
if previous is None:
@@ -168,10 +173,7 @@ def publish_due(self, now: datetime | None = None) -> int:
if token <= previous:
continue
self._last_tokens[key] = token
- closed_at = datetime.fromtimestamp(
- (token + 1) * frequency_seconds(key.timeframe),
- tz=timezone.utc,
- )
+ closed_at = completed_bar_close(key.timeframe, token, venue=key.venue, market_type=key.market_type)
session_key = (key.market, closed_at)
actionable = session_boundaries.get(session_key)
if actionable is None:
@@ -237,6 +239,8 @@ def _run(self) -> None:
key.timeframe,
self._now(),
grace_seconds=self.grace_seconds,
+ venue=key.venue,
+ market_type=key.market_type,
)
for key in keys
)
@@ -256,8 +260,11 @@ def supports_bar_close_events(frequency: str, candidates: list[dict[str, object]
):
return False
has_equity = any(
- str(item.get("underlying_market") or item.get("market") or "") in stock_markets
- or str(item.get("api_family") or "").strip().lower() == "stock"
+ equity_data_market(
+ str(item.get("market") or ""),
+ item,
+ str(item.get("api_family") or ""),
+ ) in stock_markets
for item in candidates
)
return not (has_equity and frequency_seconds(frequency) >= 86_400)
diff --git a/backend_api_python/app/events/bar_recovery.py b/backend_api_python/app/events/bar_recovery.py
new file mode 100644
index 000000000..3d66132e5
--- /dev/null
+++ b/backend_api_python/app/events/bar_recovery.py
@@ -0,0 +1,228 @@
+"""Independent recovery publisher for durable strategy bar subscriptions."""
+
+from __future__ import annotations
+
+import os
+import threading
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any, Callable
+
+from app.events.kafka import KafkaEventProducer
+from app.events.protocol import EventEnvelope, MarketBarClosedV1
+from app.services.market_schedule import is_equity_bar_close
+from app.services.strategy_event_subscriptions import StrategyEventSubscriptionRepository
+from app.services.strategy_runtime.timeframes import completed_bar_token, completed_bar_close
+from app.services.strategy_v2.frequencies import frequency_seconds, normalize_frequency
+from app.utils.logger import get_logger
+
+
+logger = get_logger(__name__)
+
+_CANONICAL_MARKETS = {
+ "astock": "AStock",
+ "cnstock": "CNStock",
+ "crypto": "Crypto",
+ "cryptocurrency": "Cryptocurrency",
+ "hkstock": "HKStock",
+ "usstock": "USStock",
+}
+
+
+@dataclass(frozen=True, slots=True)
+class BarRecoveryRoute:
+ partition_key: str
+ venue: str
+ market: str
+ market_type: str
+ instrument_id: str
+ timeframe: str
+
+ @classmethod
+ def parse(cls, partition_key: object, timeframe: object = "") -> "BarRecoveryRoute":
+ normalized = str(partition_key or "").strip()
+ parts = normalized.split(":")
+ if len(parts) < 6 or parts[0].lower() != "bar":
+ raise ValueError("strategyBarRecovery.invalidPartitionKey")
+ route_timeframe = normalize_frequency(str(timeframe or parts[-1]).strip())
+ if normalize_frequency(parts[-1]) != route_timeframe:
+ raise ValueError("strategyBarRecovery.timeframeMismatch")
+ instrument_id = ":".join(parts[4:-1]).strip()
+ if not instrument_id:
+ raise ValueError("strategyBarRecovery.instrumentRequired")
+ market_part = parts[2].strip().lower()
+ return cls(
+ partition_key=normalized,
+ venue=parts[1].strip().lower(),
+ market=_CANONICAL_MARKETS.get(market_part, parts[2].strip()),
+ market_type=parts[3].strip().lower(),
+ instrument_id=instrument_id,
+ timeframe=route_timeframe,
+ )
+
+ def event(self, token: int) -> EventEnvelope:
+ closed_at = completed_bar_close(self.timeframe, int(token), venue=self.venue, market_type=self.market_type)
+ return MarketBarClosedV1.create(
+ venue=self.venue,
+ market=self.market,
+ market_type=self.market_type,
+ instrument_id=self.instrument_id,
+ symbol=self.instrument_id,
+ timeframe=self.timeframe,
+ closed_bar_token=int(token),
+ closed_at=closed_at,
+ producer="strategy-bar-recovery",
+ )
+
+
+class StrategyBarRecoveryPublisher:
+ """Republish the latest due bar when the primary clock or mirror misses it."""
+
+ def __init__(
+ self,
+ *,
+ repository: StrategyEventSubscriptionRepository | None = None,
+ producer: KafkaEventProducer | None = None,
+ grace_seconds: float | None = None,
+ now: Callable[[], datetime] | None = None,
+ ) -> None:
+ self.repository = repository or StrategyEventSubscriptionRepository()
+ self.producer = producer or KafkaEventProducer()
+ try:
+ configured_grace = float(
+ grace_seconds
+ if grace_seconds is not None
+ else os.getenv("STRATEGY_BAR_RECOVERY_GRACE_SEC", "5")
+ )
+ except (TypeError, ValueError):
+ configured_grace = 5.0
+ self.grace_seconds = max(0.0, configured_grace)
+ self._now = now or (lambda: datetime.now(timezone.utc))
+ self._lock = threading.Lock()
+ self._last_tokens: dict[str, int] = {}
+ self._routes = 0
+ self._published = 0
+ self._publish_failures = 0
+ self._invalid_routes = 0
+ self._session_skips = 0
+
+ def observe(self, event: EventEnvelope) -> None:
+ if event.event_type != MarketBarClosedV1.event_type:
+ return
+ try:
+ token = int(event.payload.get("closed_bar_token"))
+ except (TypeError, ValueError):
+ return
+ with self._lock:
+ self._last_tokens[event.partition_key] = max(
+ token,
+ self._last_tokens.get(event.partition_key, token),
+ )
+
+ def publish_due(self, now: datetime | None = None) -> int:
+ current = now or self._now()
+ if current.tzinfo is None:
+ current = current.replace(tzinfo=timezone.utc)
+ else:
+ current = current.astimezone(timezone.utc)
+ eligible_at = current - timedelta(seconds=self.grace_seconds)
+ rows = self.repository.active_bar_routes()
+ active_keys: set[str] = set()
+ events: list[EventEnvelope] = []
+ event_tokens: dict[str, int] = {}
+ session_only_tokens: dict[str, int] = {}
+ invalid_routes = 0
+ session_skips = 0
+
+ for row in rows:
+ partition_key = str(row.get("partition_key") or "").strip()
+ if not partition_key:
+ continue
+ active_keys.add(partition_key)
+ try:
+ route = BarRecoveryRoute.parse(
+ partition_key,
+ row.get("timeframe"),
+ )
+ token = completed_bar_token(route.timeframe, eligible_at, venue=route.venue, market_type=route.market_type)
+ except Exception as exc:
+ invalid_routes += 1
+ logger.warning(
+ "Strategy bar recovery ignored invalid route: partition=%s error=%s",
+ partition_key,
+ exc,
+ )
+ continue
+ with self._lock:
+ previous = self._last_tokens.get(partition_key)
+ if previous is None:
+ previous = token - 1
+ self._last_tokens[partition_key] = previous
+ if previous is not None and token <= previous:
+ continue
+ route_has_events = False
+ for due_token in range(previous + 1, token + 1):
+ event = route.event(due_token)
+ if event.partition_key != partition_key:
+ invalid_routes += 1
+ logger.warning(
+ "Strategy bar recovery route changed during reconstruction: stored=%s generated=%s",
+ partition_key,
+ event.partition_key,
+ )
+ route_has_events = False
+ break
+ closed_at = datetime.fromisoformat(
+ str(event.payload["closed_at"]).replace("Z", "+00:00")
+ )
+ if not is_equity_bar_close(route.market, closed_at):
+ session_skips += 1
+ continue
+ events.append(event)
+ route_has_events = True
+ if route_has_events:
+ event_tokens[partition_key] = token
+ else:
+ session_only_tokens[partition_key] = token
+
+ with self._lock:
+ self._routes = len(active_keys)
+ self._invalid_routes += invalid_routes
+ self._session_skips += session_skips
+ self._last_tokens.update(session_only_tokens)
+ for stale in set(self._last_tokens) - active_keys:
+ self._last_tokens.pop(stale, None)
+
+ if not events:
+ return 0
+ if not self.producer.publish_batch(events, timeout=10.0):
+ with self._lock:
+ self._publish_failures += 1
+ logger.warning(
+ "Strategy bar recovery batch was not delivered; routes=%s",
+ len(events),
+ )
+ return 0
+ with self._lock:
+ self._last_tokens.update(event_tokens)
+ self._published += len(events)
+ return len(events)
+
+ def snapshot(self) -> dict[str, int]:
+ with self._lock:
+ return {
+ "bar_recovery_routes": self._routes,
+ "bar_recovery_published": self._published,
+ "bar_recovery_publish_failures": self._publish_failures,
+ "bar_recovery_invalid_routes": self._invalid_routes,
+ "bar_recovery_session_skips": self._session_skips,
+ }
+
+ def close(self) -> None:
+ self.producer.close()
+
+
+__all__ = [
+ "BarRecoveryRoute",
+ "StrategyBarRecoveryPublisher",
+]
diff --git a/backend_api_python/app/services/ai_generation_contracts.py b/backend_api_python/app/services/ai_generation_contracts.py
index 41779040a..83e2b5f77 100644
--- a/backend_api_python/app/services/ai_generation_contracts.py
+++ b/backend_api_python/app/services/ai_generation_contracts.py
@@ -87,7 +87,7 @@
## Orders and positions
- Order-helper signatures are exact: `order(symbol, amount)`, `order_value(symbol, value)`, `order_target(symbol, amount)`, `order_target_value(symbol, value)`, and `order_target_percent(symbol, percent)`.
- These are runtime-bound global helpers. Never pass `context` as their first argument. Optional execution and protection values must be keyword arguments after the two required arguments.
-- `get_position(symbol)` returns a `Position` object; swap code adds the optional keyword `position_side`. Read `position.amount`, `position.avg_cost`, and `position.last_price` directly; never use dictionary membership, subscripting, `.get(...)`, or `getattr(...)` on it.
+- `get_position(symbol)` returns a `Position` object. For Crypto swaps, omit `position_side` only in `one_way` mode; pass the matching `position_side="long"` or `position_side="short"` in `long_only`, `short_only`, `both`, or `neutral` modes. Read `position.amount`, `position.avg_cost`, and `position.last_price` directly; never use dictionary membership, subscripting, `.get(...)`, or `getattr(...)` on it.
- A `Position` has no `.quantity` or `.cost_basis`; use `.amount` and `.avg_cost`.
- Use `get_positions(...)` when a dictionary of multiple positions is required.
- A string argument to `get_position` returns a Position even when flat (`amount == 0`); omitted or iterable symbols return a dictionary. For hedged positions, `get_positions()` keys include `::long`/`::short`; read one leg with `get_position(symbol, position_side=...)`.
@@ -187,7 +187,7 @@
- Order helpers return `None` unless `client_order_id` is supplied. Retain explicit references for status checks; neither a reference nor a cancel request proves a completed fill.
- Use `data.current(symbol, field="close", frequency=None)` for a current scalar field. Replace every `get_current_data` call; that API does not exist.
- Enforce exact order signatures such as `order_target_percent(symbol, percent)` and never pass `context` to a global order helper.
-- Treat `get_position(symbol)` as a `Position` object with direct `.amount`, `.avg_cost`, and `.last_price` attributes; swap code adds `position_side` as a keyword. Never treat it as a dictionary or use `getattr`.
+- Treat `get_position(symbol)` as a `Position` object with direct `.amount`, `.avg_cost`, and `.last_price` attributes. For Crypto swaps, omit `position_side` only in `one_way` mode and pass the explicit long or short leg in every directional/hedged mode. Never treat it as a dictionary or use `getattr`.
- Replace legacy `.quantity` and `.cost_basis` position access with `.amount` and `.avg_cost`.
- Preserve completed-data-only execution and remove look-ahead.
- Do not remove the final history row as a candle-close workaround; the runtime supplies completed bars and keeps real-time prices outside the strategy OHLC frame.
diff --git a/backend_api_python/app/services/market_schedule.py b/backend_api_python/app/services/market_schedule.py
index 3c6e5d9d8..f4c1d836b 100644
--- a/backend_api_python/app/services/market_schedule.py
+++ b/backend_api_python/app/services/market_schedule.py
@@ -45,6 +45,139 @@ def equity_daily_bar_cutoff(market: str, now: Optional[datetime] = None) -> pd.T
return day + pd.Timedelta(days=1) - pd.Timedelta(microseconds=1)
+def _equity_week_sessions(market: str, label):
+ day = pd.Timestamp(label).tz_localize(None).normalize()
+ monday = day - pd.Timedelta(days=day.weekday())
+ return _calendar(market).sessions_in_range(monday, monday + pd.Timedelta(days=6))
+
+
+def equity_weekly_bar_cutoff(market: str, now: Optional[datetime] = None) -> pd.Timestamp:
+ current = _utc(now)
+ session = latest_completed_session(market, current, data_delay_minutes=15)
+ sessions = _equity_week_sessions(market, session)
+ if _python_utc(_calendar(market).session_close(sessions[-1])) + timedelta(minutes=15) > current:
+ sessions = _equity_week_sessions(market, session - timedelta(days=7))
+ label = pd.Timestamp(sessions[-1]).tz_localize(None).normalize()
+ if market == "HKStock":
+ label = label.tz_localize("Asia/Hong_Kong").tz_convert("UTC").tz_localize(None)
+ return label + pd.Timedelta(days=1) - pd.Timedelta(microseconds=1)
+
+
+def equity_intraday_bar_open(market: str, seconds: int, now: Optional[datetime] = None) -> pd.Timestamp:
+ """Find the last completed regular-session bar, including shortened tails."""
+ current = _utc(now)
+ calendar = _calendar(market)
+ sessions = calendar.sessions_in_range(
+ pd.Timestamp((current - timedelta(days=20)).date()), pd.Timestamp(current.date())
+ )
+ for session in reversed(sessions):
+ opens, closes = _python_utc(calendar.session_open(session)), _python_utc(calendar.session_close(session))
+ breaks = calendar.session_break_start(session), calendar.session_break_end(session)
+ segments = [(opens, closes)] if pd.isna(breaks[0]) else [
+ (opens, _python_utc(breaks[0])), (_python_utc(breaks[1]), closes)
+ ]
+ for segment_start, end in reversed(segments):
+ # Provider intraday grids remain anchored to the session open,
+ # including buckets that overlap the end of the lunch break.
+ start = opens
+ if current >= end:
+ index = max(0, int(((end - start).total_seconds() - 1) // seconds))
+ elif current >= start + timedelta(seconds=seconds):
+ index = int((current - start).total_seconds() // seconds) - 1
+ else:
+ continue
+ bar_start = start + timedelta(seconds=index * seconds)
+ if bar_start + timedelta(seconds=seconds) <= segment_start:
+ continue
+ bar_end = bar_start + timedelta(seconds=seconds)
+ closed_at = max(min(bar_end, b) for a, b in segments if bar_start < b and bar_end > a)
+ if closed_at > current:
+ continue
+ return pd.Timestamp(bar_start).tz_localize(None)
+ raise MarketScheduleError("portfolio.noCompletedMarketSession")
+
+
+def filter_closed_equity_klines(rows, timeframe: str, market: str, now: Optional[datetime] = None):
+ """Use exchange sessions, not label plus 24 hours, to confirm equity bars."""
+ from app.data_sources.base import timeframe_seconds
+ current = _utc(now)
+ seconds = timeframe_seconds(timeframe)
+ calendar = _calendar(market)
+ accepted = {}
+ for row in rows or []:
+ stamp = pd.Timestamp(int(row["time"]), unit="ms" if int(row["time"]) > 10_000_000_000 else "s", tz="UTC")
+ day = pd.Timestamp(equity_bar_session_date(stamp, market))
+ if seconds >= 604800:
+ sessions = _equity_week_sessions(market, day)
+ if not len(sessions):
+ continue
+ closes = _python_utc(calendar.session_close(sessions[-1])) + timedelta(minutes=15)
+ elif seconds >= 86400:
+ if not calendar.is_session(day):
+ continue
+ closes = _python_utc(calendar.session_close(day)) + timedelta(minutes=15)
+ else:
+ if not calendar.is_session(day):
+ continue
+ opens, end = calendar.session_open(day), calendar.session_close(day)
+ break_start, break_end = calendar.session_break_start(day), calendar.session_break_end(day)
+ segments = [(opens, end)] if pd.isna(break_start) else [(opens, break_start), (break_end, end)]
+ bar_end = stamp + pd.Timedelta(seconds=seconds)
+ intersections = [min(bar_end, b) for a, b in segments if stamp < b and bar_end > a]
+ if not intersections:
+ continue
+ closes = _python_utc(max(intersections))
+ if closes <= current:
+ accepted[row["time"]] = row
+ return [accepted[key] for key in sorted(accepted)]
+
+
+def resample_equity_klines(rows, market: str, source_seconds: int, target_seconds: int):
+ """Aggregate complete calendar buckets using only expected trading intervals."""
+ calendar = _calendar(market)
+ buckets = {}
+ expected = {}
+ for row in rows or []:
+ stamp = pd.Timestamp(int(row["time"]), unit="s", tz="UTC")
+ day = pd.Timestamp(equity_bar_session_date(stamp, market))
+ if not calendar.is_session(day):
+ continue
+ if target_seconds == 604800 and source_seconds == 86400:
+ sessions = _equity_week_sessions(market, day)
+ start = day - pd.Timedelta(days=day.weekday())
+ key = int(start.tz_localize(calendar.tz).timestamp())
+ expected[key] = {session.date() for session in sessions}
+ source_key = day.date()
+ else:
+ opens, closes = calendar.session_open(day), calendar.session_close(day)
+ break_start, break_end = calendar.session_break_start(day), calendar.session_break_end(day)
+ segments = [(opens, closes)] if pd.isna(break_start) else [(opens, break_start), (break_end, closes)]
+ source_end = stamp + pd.Timedelta(seconds=source_seconds)
+ if stamp < opens or not any(stamp < b and source_end > a for a, b in segments):
+ continue
+ offset = int((stamp - opens).total_seconds()) // target_seconds * target_seconds
+ start = opens + pd.Timedelta(seconds=offset)
+ end = min(start + pd.Timedelta(seconds=target_seconds), closes)
+ key = int(start.timestamp())
+ expected[key] = {
+ value for value in range(key, int(end.timestamp()), source_seconds)
+ if any(value < int(b.timestamp()) and value + source_seconds > int(a.timestamp()) for a, b in segments)
+ }
+ source_key = int(stamp.timestamp())
+ buckets.setdefault(key, {})[source_key] = row
+ merged = []
+ for key, bucket in sorted(buckets.items()):
+ if not expected[key].issubset(bucket):
+ continue
+ chunk = [bucket[item] for item in sorted(expected[key])]
+ merged.append({
+ "time": key, "open": chunk[0]["open"], "close": chunk[-1]["close"],
+ "high": max(row["high"] for row in chunk), "low": min(row["low"] for row in chunk),
+ "volume": round(sum(row.get("volume", 0) for row in chunk), 2),
+ })
+ return merged
+
+
def equity_daily_execution_session(market: str, regular_only: bool = True, now: Optional[datetime] = None) -> Optional[datetime]:
"""Use the next regular open for brokers, or completed reference data for venues."""
current = _utc(now)
diff --git a/backend_api_python/app/services/strategy_event_subscriptions.py b/backend_api_python/app/services/strategy_event_subscriptions.py
index 5a4f767e0..6673110b1 100644
--- a/backend_api_python/app/services/strategy_event_subscriptions.py
+++ b/backend_api_python/app/services/strategy_event_subscriptions.py
@@ -22,6 +22,35 @@ def strategy_shard(strategy_id: int, shard_count: int | None = None) -> int:
class StrategyEventSubscriptionRepository:
+ def active_bar_routes(self) -> list[dict[str, str]]:
+ """Return distinct bar routes still referenced by running strategies."""
+ with get_db_connection() as db:
+ cur = db.cursor()
+ try:
+ cur.execute(
+ """
+ SELECT DISTINCT subscription.partition_key, subscription.timeframe
+ FROM qd_strategy_event_subscriptions AS subscription
+ JOIN qd_strategies_trading AS strategy
+ ON strategy.id = subscription.strategy_id
+ WHERE subscription.event_type = %s
+ AND strategy.status = 'running'
+ ORDER BY subscription.partition_key
+ """,
+ (MarketBarClosedV1.event_type,),
+ )
+ rows: list[dict[str, Any]] = cur.fetchall() or []
+ finally:
+ cur.close()
+ return [
+ {
+ "partition_key": str(row.get("partition_key") or ""),
+ "timeframe": str(row.get("timeframe") or ""),
+ }
+ for row in rows
+ if str(row.get("partition_key") or "").strip()
+ ]
+
def replace_bar_streams(
self,
strategy_id: int,
diff --git a/backend_api_python/app/services/strategy_runtime/candle_readiness.py b/backend_api_python/app/services/strategy_runtime/candle_readiness.py
new file mode 100644
index 000000000..ea9efc4f1
--- /dev/null
+++ b/backend_api_python/app/services/strategy_runtime/candle_readiness.py
@@ -0,0 +1,88 @@
+"""Keep live signal progress behind confirmed, current market data."""
+
+from datetime import datetime, timedelta, timezone
+
+import pandas as pd
+
+from app.data_sources.candle_calendar import completed_candle_open
+from app.data_sources.base import timeframe_seconds
+from app.services.market_schedule import (
+ equity_data_market, equity_bar_session_date, equity_daily_bar_cutoff,
+ equity_weekly_bar_cutoff, equity_intraday_bar_open,
+)
+
+
+def expected_bar(member, frequency, now=None, frame=None):
+ current = now or datetime.now(timezone.utc)
+ if current.tzinfo is None:
+ current = current.replace(tzinfo=timezone.utc)
+ market = equity_data_market(str(member.get("market") or "Crypto"), member)
+ seconds = timeframe_seconds(frequency)
+ if market in {"USStock", "HKStock"}:
+ if seconds == 86400:
+ return equity_daily_bar_cutoff(market, current)
+ if seconds == 604800:
+ return equity_weekly_bar_cutoff(market, current)
+ return equity_intraday_bar_open(market, seconds, current)
+ venue = str(member.get("exchange_id") or "")
+ if not venue and market.lower() in {"crypto", "cryptocurrency"} and frame is not None and not frame.empty:
+ origin = int(pd.Timestamp(frame.index[-1]).timestamp()) % seconds
+ stamp = int((current.timestamp() - origin) // seconds) * seconds + origin - seconds
+ else:
+ stamp = completed_candle_open(current.timestamp(), frequency, venue, str(member.get("market_type") or "spot"))
+ return pd.Timestamp(datetime.fromtimestamp(stamp, timezone.utc)).tz_localize(None)
+
+
+def frame_is_current(frame, member, frequency, now=None):
+ if frame is None or frame.empty:
+ return False
+ expected = expected_bar(member, frequency, now, frame)
+ return frame_covers_bar(frame, member, frequency, expected)
+
+
+def frame_covers_bar(frame, member, frequency, expected):
+ if frame is None or frame.empty:
+ return False
+ latest = pd.Timestamp(frame.index[-1])
+ if latest.tzinfo is not None:
+ latest = latest.tz_convert("UTC").tz_localize(None)
+ market = equity_data_market(str(member.get("market") or "Crypto"), member)
+ if market in {"USStock", "HKStock"} and timeframe_seconds(frequency) >= 86400:
+ actual_day = equity_bar_session_date(latest, market)
+ expected_day = equity_bar_session_date(expected, market)
+ if timeframe_seconds(frequency) == 604800:
+ return actual_day.isocalendar()[:2] == expected_day.isocalendar()[:2]
+ return actual_day == expected_day
+ return latest == expected
+
+
+def signal_bar_targets(candidates, frequency, frames, now=None):
+ current = now or datetime.now(timezone.utc)
+ return tuple(
+ (str(member["key"]), str(expected_bar(member, frequency, current, frames.get(str(member["key"])))))
+ for member in candidates
+ )
+
+
+def runtime_frames_ready(bundles, candidates, now=None):
+ if not bundles or not candidates:
+ return False
+ current = now or datetime.now(timezone.utc)
+ # A clock boundary is not a provider finality acknowledgement. Delay
+ # evaluation briefly, without advancing the processed-bar watermark.
+ eligible = current - timedelta(seconds=2)
+ for frequency, frames in bundles.items():
+ for member in candidates:
+ frame = frames.get(str(member["key"]))
+ if not frame_is_current(frame, member, frequency, current):
+ return False
+ if expected_bar(member, frequency, eligible, frame) != expected_bar(member, frequency, current, frame):
+ return False
+ return True
+
+
+def candles_settling(candidates, frequency, frames, now=None):
+ current = now or datetime.now(timezone.utc)
+ return signal_bar_targets(candidates, frequency, frames, current) != signal_bar_targets(
+ candidates, frequency, frames, current - timedelta(seconds=2)
+ )
diff --git a/backend_api_python/app/services/strategy_runtime/timeframes.py b/backend_api_python/app/services/strategy_runtime/timeframes.py
index 1851b57ad..91e5f2555 100644
--- a/backend_api_python/app/services/strategy_runtime/timeframes.py
+++ b/backend_api_python/app/services/strategy_runtime/timeframes.py
@@ -56,13 +56,23 @@ def live_history_days(
return days
-def completed_bar_token(frequency: str, now: datetime | None = None) -> int:
+def completed_bar_token(frequency: str, now: datetime | None = None, *, venue: str = "", market_type: str = "spot") -> int:
"""Return a stable token for the latest fully completed UTC candle."""
current = now or datetime.now(timezone.utc)
if current.tzinfo is None:
current = current.replace(tzinfo=timezone.utc)
seconds = max(1, frequency_seconds(frequency))
- return int(current.timestamp()) // seconds - 1
+ from app.data_sources.candle_calendar import candle_origin
+ origin = candle_origin(frequency, venue, market_type)
+ return int((current.timestamp() - origin) // seconds) - 1
+
+
+def completed_bar_close(frequency: str, token: int, *, venue: str = "", market_type: str = "spot") -> datetime:
+ from app.data_sources.candle_calendar import candle_origin
+ return datetime.fromtimestamp(
+ (token + 1) * frequency_seconds(frequency) + candle_origin(frequency, venue, market_type),
+ timezone.utc,
+ )
def seconds_until_next_completed_bar(
@@ -70,6 +80,8 @@ def seconds_until_next_completed_bar(
now: datetime | None = None,
*,
grace_seconds: float = 0.25,
+ venue: str = "",
+ market_type: str = "spot",
) -> float:
"""Return the delay until the next UTC candle can be treated as closed."""
current = now or datetime.now(timezone.utc)
@@ -77,7 +89,9 @@ def seconds_until_next_completed_bar(
current = current.replace(tzinfo=timezone.utc)
seconds = max(1, frequency_seconds(frequency))
timestamp = current.timestamp()
- next_boundary = (math.floor(timestamp / seconds) + 1) * seconds
+ from app.data_sources.candle_calendar import candle_origin
+ origin = candle_origin(frequency, venue, market_type)
+ next_boundary = (math.floor((timestamp - origin) / seconds) + 1) * seconds + origin
return max(0.0, next_boundary - timestamp + max(0.0, float(grace_seconds or 0.0)))
diff --git a/backend_api_python/app/services/strategy_v2/contract.py b/backend_api_python/app/services/strategy_v2/contract.py
index 1dc3ed452..2bd977458 100644
--- a/backend_api_python/app/services/strategy_v2/contract.py
+++ b/backend_api_python/app/services/strategy_v2/contract.py
@@ -559,6 +559,26 @@ def canonical_source_metadata(
"order_target_percent": ("symbol", "percent"),
}
+_ORDER_API_OPTION_NAMES = {
+ "client_order_id",
+ "execution_algo",
+ "limit_price",
+ "maker_offset_bps",
+ "maker_wait_sec",
+ "order_type",
+ "position_side",
+ "price",
+ "protection",
+ "reason",
+ "stop_loss_pct",
+ "take_profit_pct",
+ "time_limit_seconds",
+ "trailing_activation_pct",
+ "trailing_rebase_on_scale_in",
+ "trailing_stop_pct",
+ "type",
+}
+
_UNSUPPORTED_STRATEGY_API_NAMES = {
"get_current_data": "data.current",
}
@@ -931,6 +951,11 @@ def _validate_order_call(
) -> None:
required = _ORDER_API_ARGUMENTS[call_name]
keywords = _call_keywords(node)
+ unsupported = sorted(set(keywords) - set(required) - _ORDER_API_OPTION_NAMES)
+ if unsupported:
+ raise StrategyV2ContractError(
+ f"strategyV2.apiCallInvalid:{call_name}:unsupportedArgument:{unsupported[0]}"
+ )
if len(node.args) > 2:
raise StrategyV2ContractError(
f"strategyV2.apiCallInvalid:{call_name}:expectedSymbolAndValue"
diff --git a/backend_api_python/app/services/strategy_v2/market_data.py b/backend_api_python/app/services/strategy_v2/market_data.py
index 2f0e4a093..c15395695 100644
--- a/backend_api_python/app/services/strategy_v2/market_data.py
+++ b/backend_api_python/app/services/strategy_v2/market_data.py
@@ -94,11 +94,22 @@ def _last_completed_bar_open(
).tz_localize(None)
-def _market_bar_cutoff(market: str, timeframe: str, *, now=None) -> pd.Timestamp:
+def _market_bar_cutoff(market: str, timeframe: str, *, now=None, exchange_id=None, market_type=None) -> pd.Timestamp:
if market in {"USStock", "HKStock"} and timeframe == "1d":
from app.services.market_schedule import equity_daily_bar_cutoff
return equity_daily_bar_cutoff(market, now)
+ if market in {"USStock", "HKStock"} and timeframe == "1w":
+ from app.services.market_schedule import equity_weekly_bar_cutoff
+ return equity_weekly_bar_cutoff(market, now)
+ if market in {"USStock", "HKStock"} and TIMEFRAME_SECONDS.get(timeframe, 86400) < 86400:
+ from app.services.market_schedule import equity_intraday_bar_open
+ return equity_intraday_bar_open(market, TIMEFRAME_SECONDS[timeframe], now)
+ if market.lower() in {"crypto", "cryptocurrency"}:
+ from app.data_sources.candle_calendar import candle_origin
+ current = _normalize_utc_datetime(now or datetime.now(timezone.utc))
+ origin = candle_origin(timeframe, exchange_id or "", market_type or "spot")
+ return _last_completed_bar_open(TIMEFRAME_SECONDS[timeframe], now=current - timedelta(seconds=origin)) + pd.Timedelta(seconds=origin)
return _last_completed_bar_open(TIMEFRAME_SECONDS.get(timeframe, 86400), now=now)
@@ -172,7 +183,7 @@ def _load_strategy_frame_uncached(
timeframe_seconds = TIMEFRAME_SECONDS.get(normalized_timeframe, 86400)
provider_timeframe = PROVIDER_TIMEFRAMES.get(normalized_timeframe, normalized_timeframe)
data_market = equity_data_market(market, product, resolved_api_family)
- closed_bar_cutoff = _market_bar_cutoff(data_market, normalized_timeframe)
+ closed_bar_cutoff = _market_bar_cutoff(data_market, normalized_timeframe, exchange_id=exchange_id, market_type=market_type)
cutoff_utc = closed_bar_cutoff.to_pydatetime().replace(tzinfo=timezone.utc)
provider_end_utc = min(end_utc, cutoff_utc)
total_seconds = max(1.0, (provider_end_utc - start_utc).total_seconds())
@@ -180,7 +191,7 @@ def _load_strategy_frame_uncached(
after_time = int((start_utc - timedelta(seconds=timeframe_seconds)).timestamp())
before_time = int((provider_end_utc + timedelta(seconds=timeframe_seconds)).timestamp())
cache_key = ":".join((
- "session-calendar-v2",
+ "confirmed-candle-calendar-v3",
str(market),
str(symbol),
str(timeframe),
@@ -236,6 +247,11 @@ def _load_strategy_frame_uncached(
coverage_end,
timeframe_seconds,
)
+ if cache_is_usable and requested_end >= closed_bar_cutoff and data_market in {"Crypto", "Cryptocurrency", "USStock", "HKStock"}:
+ from app.services.strategy_runtime.candle_readiness import frame_covers_bar
+ cache_is_usable = frame_covers_bar(
+ cached, {"market": data_market}, normalized_timeframe, closed_bar_cutoff,
+ )
if cache_is_usable:
return cached.copy()
logger.warning(
@@ -624,7 +640,9 @@ def load_strategy_frame(
market, symbol, exchange_id=exchange_id, market_type=market_type, instrument_id=instrument_id,
)
data_market = equity_data_market(market, product, str(api_family or ""))
- closed_cutoff = _market_bar_cutoff(data_market, normalized_timeframe)
+ from app.services.strategy_runtime.candle_readiness import frame_covers_bar
+ readiness_member = dict(product or {}, market=data_market, exchange_id=exchange_id or "", market_type=market_type or "spot")
+ closed_cutoff = _market_bar_cutoff(data_market, normalized_timeframe, exchange_id=exchange_id, market_type=market_type)
coverage_end = min(requested_end, closed_cutoff)
live_request = _is_live_request(
requested_end,
@@ -679,6 +697,10 @@ def load_strategy_frame(
coverage_end,
timeframe_seconds,
)
+ if live_request and data_market in {"USStock", "HKStock"}:
+ cached_result_is_usable = cached_result_is_usable and frame_covers_bar(
+ cached_result, readiness_member, normalized_timeframe, closed_cutoff,
+ )
if cached_result_is_usable:
entry.last_access = time.monotonic()
return cached_result
@@ -754,10 +776,10 @@ def load_strategy_frame(
merged = merged[merged.index >= requested_start - overlap]
actual_start = merged.index.min()
actual_end = merged.index.max()
- incomplete_equity_daily = (
- data_market in {"USStock", "HKStock"} and normalized_timeframe == "1d"
+ incomplete_equity_window = (
+ data_market in {"USStock", "HKStock"}
and live_request
- and equity_bar_session_date(actual_end, data_market) < equity_bar_session_date(coverage_end, data_market)
+ and not frame_covers_bar(merged, readiness_member, normalized_timeframe, closed_cutoff)
)
complete_equity_window = bool(
equity_market
@@ -777,7 +799,7 @@ def load_strategy_frame(
)
cached_coverage_end = (
coverage_end
- if complete_equity_window and not incomplete_equity_daily
+ if complete_equity_window and not incomplete_equity_window
else actual_end
if continuous_crypto_market or equity_market
else max(actual_end, coverage_end)
diff --git a/backend_api_python/app/services/strategy_v2/runtime.py b/backend_api_python/app/services/strategy_v2/runtime.py
index 0f9e3d7ab..1fee7ec2b 100644
--- a/backend_api_python/app/services/strategy_v2/runtime.py
+++ b/backend_api_python/app/services/strategy_v2/runtime.py
@@ -47,6 +47,26 @@
"time_limit",
})
+_ORDER_OPTION_NAMES = frozenset({
+ "client_order_id",
+ "execution_algo",
+ "limit_price",
+ "maker_offset_bps",
+ "maker_wait_sec",
+ "order_type",
+ "position_side",
+ "price",
+ "protection",
+ "reason",
+ "stop_loss_pct",
+ "take_profit_pct",
+ "time_limit_seconds",
+ "trailing_activation_pct",
+ "trailing_rebase_on_scale_in",
+ "trailing_stop_pct",
+ "type",
+})
+
def _backtest_time_iso(value: Any) -> str:
"""Serialize the UTC-naive market index as an unambiguous UTC instant."""
@@ -373,6 +393,25 @@ def get_position(
current = self.portfolio.positions.get(leg_key)
if current is not None:
return current
+ if not _normalize_position_side(position_side):
+ leg_prefix = f"{key}::"
+ matching_legs = [
+ position
+ for stored_key, position in self.portfolio.positions.items()
+ if stored_key.startswith(leg_prefix)
+ and _normalize_position_side(stored_key[len(leg_prefix):])
+ ]
+ open_legs = [
+ position
+ for position in matching_legs
+ if abs(float(position.amount or 0.0)) > 1e-12
+ ]
+ if len(open_legs) == 1:
+ return open_legs[0]
+ if len(open_legs) > 1:
+ raise StrategyV2ContractError("strategyV2.positionSideRequired")
+ if len(matching_legs) == 1:
+ return matching_legs[0]
try:
last_price = self.portal.current(key, "close", 0.0)
except Exception:
@@ -589,6 +628,11 @@ def _queue(
value: object,
kwargs: Mapping[str, Any],
) -> str | None:
+ unsupported = sorted(set(kwargs) - _ORDER_OPTION_NAMES)
+ if unsupported:
+ raise StrategyV2ContractError(
+ f"strategyV2.apiCallInvalid:{kind}:unsupportedArgument:{unsupported[0]}"
+ )
key = self.portal.resolve_key(symbol)
try:
number = float(value)
diff --git a/backend_api_python/app/services/trading_executor.py b/backend_api_python/app/services/trading_executor.py
index fa9c24d44..1212dc4e1 100644
--- a/backend_api_python/app/services/trading_executor.py
+++ b/backend_api_python/app/services/trading_executor.py
@@ -1059,13 +1059,14 @@ def wake_for_closed_bar(event) -> None:
)
next_signal_poll = 0.0
next_position_mark_at = 0.0
- last_signal_bar_token: int | None = None
+ last_signal_bar_token = None
last_processed_frame_timestamp: pd.Timestamp | None = None
initial_frames_pending = True
stale_price_logged = False
consecutive_errors = 0
last_market_data_failure_key = ""
last_market_data_log_at = 0.0
+ signal_data_error = ""
strategy_name = str(strategy.get("strategy_name") or f"strategy_{strategy_id}")
notification_config = _json_object(strategy.get("notification_config"))
leverage = max(1.0, float(trading_config.get("leverage") or strategy.get("leverage") or 1))
@@ -1358,20 +1359,39 @@ def wake_for_closed_bar(event) -> None:
from app.services.market_schedule import equity_daily_execution_session
signal_session = equity_daily_execution_session(*daily_policy) if daily_policy else None
- current_bar_token = completed_bar_token(frequency)
+ from app.services.strategy_runtime.candle_readiness import (
+ candles_settling, runtime_frames_ready, signal_bar_targets,
+ )
+ signal_members = [
+ member for member in candidates
+ if str(member.get("key") or "") in ready_universe_symbols
+ ]
+ current_bar_token = signal_bar_targets(signal_members, frequency, frames)
if signal_session is not None:
current_bar_token = int(signal_session.timestamp())
has_new_closed_bar = (
initial_frames_pending
or current_bar_token != last_signal_bar_token
- ) and (not daily_policy or signal_session is not None)
+ ) and (not daily_policy or signal_session is not None) and not candles_settling(signal_members, frequency, frames)
if has_new_closed_bar:
# Startup already warmed the complete frame bundle.
# Every later trigger extends the shared cache only
# for the newly completed candle window.
- if not initial_frames_pending or daily_policy:
+ if not initial_frames_pending or daily_policy or not runtime_frames_ready(frequency_frames, signal_members):
frequency_frames = fetch_runtime_frames()
frames = frequency_frames[frequency]
+ signal_members = [
+ member for member in candidates
+ if str(member.get("key") or "") in ready_universe_symbols
+ ]
+ if not runtime_frames_ready(frequency_frames, signal_members):
+ raise MarketDataUnavailableError(classify_market_data_failure(
+ "Incomplete K-line history: latest closed candle is not ready",
+ exchange_id=str(primary.get("exchange_id") or ""),
+ market_type=str(primary.get("market_type") or ""),
+ symbol=str(primary.get("symbol") or ""), timeframe=frequency,
+ ))
+ signal_data_error = ""
daily_members = [
member
for member in candidates
@@ -1501,8 +1521,8 @@ def wake_for_closed_bar(event) -> None:
last_prices,
pending_count,
loop_latency_ms=int((time.monotonic() - cycle_started) * 1000),
- status="healthy" if active_prices else "degraded",
- last_error="" if active_prices else "live_price_stale",
+ status="healthy" if active_prices and not signal_data_error else "degraded",
+ last_error=signal_data_error or ("" if active_prices else "live_price_stale"),
price_source=str(price_feed_meta.get("source") or "exchange_rest"),
price_age_ms=int(price_feed_meta.get("age_ms") or 0),
trigger_mode=(
@@ -1533,7 +1553,7 @@ def wake_for_closed_bar(event) -> None:
self._mark_stopped(strategy_id)
break
consecutive_errors = 0
- if last_market_data_failure_key:
+ if last_market_data_failure_key and not signal_data_error:
append_strategy_log(strategy_id, "info", "Market data feed recovered")
last_market_data_failure_key = ""
last_market_data_log_at = 0.0
@@ -1553,6 +1573,7 @@ def wake_for_closed_bar(event) -> None:
failure.timeframe,
failure.technical_detail,
))
+ signal_data_error = f"marketData.{failure.code}"
logger.warning(
"Strategy %s market data unavailable (%s): %s",
strategy_id,
diff --git a/backend_api_python/tests/test_candle_finality.py b/backend_api_python/tests/test_candle_finality.py
new file mode 100644
index 000000000..9c507f08d
--- /dev/null
+++ b/backend_api_python/tests/test_candle_finality.py
@@ -0,0 +1,225 @@
+"""Regression coverage for venue boundaries and exchange-session finality."""
+
+from datetime import datetime, timezone
+
+import pandas as pd
+import pytest
+
+from app.data_sources.candle_calendar import candle_open, completed_candle_open
+from app.data_sources.native_crypto import NativeCryptoPublicClient
+from app.services.market_schedule import filter_closed_equity_klines, equity_intraday_bar_open
+from app.services.strategy_v2.market_data import _market_bar_cutoff
+
+
+def utc(value):
+ return datetime.fromisoformat(value).replace(tzinfo=timezone.utc)
+
+
+@pytest.mark.parametrize("frequency,step", [("1m", 60), ("5m", 300), ("1h", 3600), ("4h", 14400), ("1d", 86400), ("1w", 604800)])
+def test_gate_futures_reads_the_entire_final_candle(frequency, step):
+ client = NativeCryptoPublicClient("gate", "swap")
+ target = completed_candle_open(utc("2026-01-10T12:00:00").timestamp(), frequency, "gate", "swap")
+ def get(_url, params):
+ assert params["to"] == target + step - 1
+ return [{"t": target, "o": 10, "h": 12, "l": 9, "c": 11, "v": 42}]
+ client._get = get
+ rows = client.fetch_ohlcv("BTC/USDT", frequency, since=(target - 4 * step) * 1000, limit=5, until=(target + step) * 1000)
+ assert rows[-1][4:] == [11, 42]
+
+
+@pytest.mark.parametrize("market", ["spot", "swap"])
+def test_bitget_daily_end_uses_native_session_boundary(market):
+ client = NativeCryptoPublicClient("bitget", market)
+ target = int(utc("2026-01-05T16:00:00").timestamp() * 1000)
+ def get(_url, params):
+ assert params["endTime"] == target + 86400000
+ return {"data": [[target, 10, 12, 9, 11, 42]]}
+ client._get = get
+ assert client.fetch_ohlcv("BTC/USDT", "1d", since=target - 4 * 86400000, limit=5)[-1][0] == target
+
+
+@pytest.mark.parametrize("venue,market", [("okx", "spot"), ("okx", "swap"), ("gate", "spot")])
+def test_explicit_unconfirmed_candles_do_not_reach_strategy(venue, market):
+ client = NativeCryptoPublicClient(venue, market)
+ target = int(utc("2026-01-01T12:00:00").timestamp() * 1000)
+ raw = {"data": [[target, 10, 12, 9, 11, 42, 1, 1, "0"]]}
+ if venue == "gate":
+ raw = [[target // 1000, 1, 11, 12, 9, 10, 42, "false"]]
+ client._get = lambda *_args: raw
+ assert client.fetch_ohlcv("BTC/USDT", "1m", since=target, limit=1)
+ assert client.fetch_ohlcv("BTC/USDT", "1m", since=target, limit=1, closed_only=True) == []
+
+
+@pytest.mark.parametrize("venue,market,expected", [
+ ("okx", "swap", "2026-11-05T16:00:00"), ("bitget", "spot", "2026-11-05T16:00:00"),
+ ("htx", "swap", "2026-11-05T16:00:00"), ("binance", "spot", "2026-11-05T00:00:00"),
+ ("bybit", "swap", "2026-11-05T00:00:00"), ("gate", "swap", "2026-11-05T00:00:00"),
+])
+def test_daily_strategy_cutoff_matches_venue(venue, market, expected):
+ result = _market_bar_cutoff("Crypto", "1d", now=utc("2026-11-06T16:00:05"), exchange_id=venue, market_type=market)
+ assert result == pd.Timestamp(expected)
+
+
+@pytest.mark.parametrize("venue,market,expected", [
+ ("binance", "spot", "2026-10-26T00:00:00"), ("bybit", "swap", "2026-10-26T00:00:00"),
+ ("gate", "spot", "2026-10-26T00:00:00"), ("gate", "swap", "2026-10-22T00:00:00"),
+ ("okx", "swap", "2026-10-25T16:00:00"), ("htx", "spot", "2026-10-25T16:00:00"),
+ ("bitget", "swap", "2026-10-25T16:00:00"),
+])
+def test_weekly_boundary_is_not_always_epoch_thursday(venue, market, expected):
+ result = completed_candle_open(utc("2026-11-02T00:00:05").timestamp(), "1w", venue, market)
+ assert result == int(utc(expected).timestamp())
+
+
+@pytest.mark.parametrize("market,label,before,after", [
+ ("USStock", "2025-07-07T04:00:00", "2025-07-07T20:10:00", "2025-07-07T20:16:00"),
+ ("USStock", "2025-11-03T05:00:00", "2025-11-03T21:10:00", "2025-11-03T21:16:00"),
+ ("USStock", "2025-07-03T04:00:00", "2025-07-03T17:10:00", "2025-07-03T17:16:00"),
+ ("HKStock", "2025-07-06T16:00:00", "2025-07-07T08:10:00", "2025-07-07T08:16:00"),
+])
+def test_daily_equity_close_uses_session_not_24_hours(market, label, before, after):
+ rows = [{"time": int(utc(label).timestamp()), "close": 10}]
+ assert filter_closed_equity_klines(rows, "1d", market, utc(before)) == []
+ assert filter_closed_equity_klines(rows, "1d", market, utc(after)) == rows
+
+
+def test_weekly_equity_waits_for_shortened_last_session():
+ rows = [{"time": int(utc("2025-06-30T04:00:00").timestamp()), "close": 10}]
+ assert filter_closed_equity_klines(rows, "1w", "USStock", utc("2025-07-03T17:10:00")) == []
+ assert filter_closed_equity_klines(rows, "1w", "USStock", utc("2025-07-03T17:16:00")) == rows
+
+
+@pytest.mark.parametrize("market,now,expected", [
+ ("HKStock", "2025-07-07T04:05:00", "2025-07-07T03:30:00"),
+ ("HKStock", "2025-07-07T04:55:00", "2025-07-07T03:30:00"),
+ ("USStock", "2025-07-03T17:05:00", "2025-07-03T16:30:00"),
+ ("USStock", "2025-07-04T14:05:00", "2025-07-03T16:30:00"),
+])
+def test_intraday_completion_handles_breaks_holidays_and_short_sessions(market, now, expected):
+ assert equity_intraday_bar_open(market, 3600, utc(now)) == pd.Timestamp(expected)
+
+
+def frame(label):
+ return pd.DataFrame({"open": [10.], "high": [12.], "low": [9.], "close": [11.], "volume": [42.]}, index=[pd.Timestamp(label)])
+
+
+def test_runtime_waits_for_confirmed_tail_and_boundary_settling():
+ from app.services.strategy_runtime.candle_readiness import runtime_frames_ready
+ members = [{"key": "btc", "market": "Crypto", "exchange_id": "gate", "market_type": "swap"}]
+ bundle = {"1m": {"btc": frame("2026-01-01T11:59:00")}}
+ assert not runtime_frames_ready(bundle, members, utc("2026-01-01T12:01:05"))
+ bundle["1m"]["btc"] = frame("2026-01-01T12:00:00")
+ assert not runtime_frames_ready(bundle, members, utc("2026-01-01T12:01:01"))
+ assert runtime_frames_ready(bundle, members, utc("2026-01-01T12:01:05"))
+ assert not runtime_frames_ready({}, members, utc("2026-01-01T12:01:05"))
+
+
+def test_secondary_timeframe_must_also_be_current():
+ from app.services.strategy_runtime.candle_readiness import runtime_frames_ready
+ members = [{"key": "btc", "market": "Crypto", "exchange_id": "okx"}]
+ bundle = {"1m": {"btc": frame("2026-01-01T16:00:00")}, "1d": {"btc": frame("2025-12-30T16:00:00")}}
+ assert not runtime_frames_ready(bundle, members, utc("2026-01-01T16:01:05"))
+ bundle["1d"]["btc"] = frame("2025-12-31T16:00:00")
+ assert runtime_frames_ready(bundle, members, utc("2026-01-01T16:01:05"))
+
+
+def test_equity_weekend_is_current_but_stock_perpetual_is_not():
+ from app.services.strategy_runtime.candle_readiness import frame_is_current
+ weekend = utc("2025-07-05T15:00:05")
+ last_session = frame("2025-07-03T16:59:00")
+ assert frame_is_current(last_session, {"market": "USStock"}, "1m", weekend)
+ assert frame_is_current(last_session, {"market": "Crypto", "api_family": "stock", "underlying_market": "USStock"}, "1m", weekend)
+ assert not frame_is_current(last_session, {"market": "Crypto", "api_family": "futures", "underlying_market": "USStock", "market_type": "swap"}, "1m", weekend)
+
+
+@pytest.mark.parametrize("venue,frequency", [("okx", "1d"), ("bitget", "1d"), ("htx", "1w"), ("gate", "1w")])
+def test_event_token_and_recovery_close_share_native_boundary(venue, frequency):
+ from app.events.bar_recovery import BarRecoveryRoute
+ from app.services.strategy_runtime.timeframes import completed_bar_token, completed_bar_close
+ now = utc("2026-11-06T16:00:05")
+ token = completed_bar_token(frequency, now, venue=venue, market_type="swap")
+ close = completed_bar_close(frequency, token, venue=venue, market_type="swap")
+ step = 604800 if frequency == "1w" else 86400
+ assert close.timestamp() == completed_candle_open(now.timestamp(), frequency, venue, "swap") + step
+ route = BarRecoveryRoute.parse(f"bar:{venue}:crypto:swap:btc/usdt:{frequency}")
+ assert datetime.fromisoformat(route.event(token).payload["closed_at"].replace("Z", "+00:00")) == close
+
+
+def test_gate_historical_asof_inside_bar_cannot_confirm_truncated_candle():
+ client = NativeCryptoPublicClient("gate", "swap")
+ target = int(utc("2026-01-01T12:00:00").timestamp())
+ client._get = lambda *_: [{"t": target, "o": 10, "h": 10, "l": 10, "c": 10, "v": 0}]
+ assert client.fetch_ohlcv("BTC/USDT", "1m", since=target * 1000, until=(target + 20) * 1000, closed_only=True) == []
+
+
+def bars(labels):
+ return [dict(time=int(utc(label).timestamp()), open=10, high=12, low=9, close=11, volume=42) for label in labels]
+
+
+def test_hk_four_hour_resample_preserves_grid_and_does_not_join_next_day():
+ from app.services.market_schedule import resample_equity_klines
+ rows = bars(["2025-07-07T01:30:00", "2025-07-07T02:30:00", "2025-07-07T03:30:00", "2025-07-07T04:30:00", "2025-07-07T05:30:00", "2025-07-07T06:30:00", "2025-07-07T07:30:00", "2025-07-08T01:30:00"])
+ result = resample_equity_klines(rows, "HKStock", 3600, 14400)
+ assert [row["time"] for row in result] == [rows[0]["time"], rows[4]["time"]]
+ assert [row["volume"] for row in result] == [168, 126]
+ assert resample_equity_klines(rows[:1] + rows[2:3], "HKStock", 3600, 14400) == []
+
+
+def test_hk_lunch_overlap_is_not_dropped_or_prematurely_finalized():
+ row = bars(["2025-07-07T04:30:00"])
+ assert filter_closed_equity_klines(row, "1h", "HKStock", utc("2025-07-07T05:20:00")) == []
+ assert filter_closed_equity_klines(row, "1h", "HKStock", utc("2025-07-07T05:31:00")) == row
+ assert equity_intraday_bar_open("HKStock", 3600, utc("2025-07-07T05:31:00")) == pd.Timestamp("2025-07-07T04:30:00")
+ four_hour = bars(["2025-07-07T01:30:00"])
+ assert filter_closed_equity_klines(four_hour, "4h", "HKStock", utc("2025-07-07T04:10:00")) == []
+ assert equity_intraday_bar_open("HKStock", 14400, utc("2025-07-07T04:10:00")) < pd.Timestamp("2025-07-07")
+ assert filter_closed_equity_klines(four_hour, "4h", "HKStock", utc("2025-07-07T05:31:00")) == four_hour
+
+
+def test_hk_half_hour_fallback_matches_native_hour_grid():
+ from app.data_sources.asia_stock_kline import _merge_session_bars
+ rows = bars(["2025-07-07T01:30:00", "2025-07-07T02:00:00", "2025-07-07T02:30:00", "2025-07-07T03:00:00", "2025-07-07T03:30:00", "2025-07-07T05:00:00", "2025-07-07T05:30:00", "2025-07-07T06:00:00", "2025-07-07T06:30:00", "2025-07-07T07:00:00", "2025-07-07T07:30:00"])
+ result = _merge_session_bars(rows, "1H", True, source_seconds=1800)
+ assert [row["time"] for row in result] == [int(utc(f"2025-07-07T{hour:02d}:30:00").timestamp()) for hour in range(1, 8)]
+ assert sum(row["volume"] for row in result) == sum(row["volume"] for row in rows)
+
+
+def test_hk_akshare_close_labels_are_normalized_to_utc_open():
+ from app.data_sources.asia_stock_kline import _bars_from_ak_min_df
+ source = pd.DataFrame([["2025-07-07 10:00:00", 10, 11, 12, 9, 42]], columns=["time", "open", "close", "high", "low", "volume"])
+ rows = _bars_from_ak_min_df(source, is_hk=True, period_seconds=1800)
+ assert rows == bars(["2025-07-07T01:30:00"])
+
+
+def test_twelvedata_hk_uses_local_timezone_and_half_hour_constituents(monkeypatch):
+ from types import SimpleNamespace
+ from app.data_sources import asia_stock_kline
+ monkeypatch.setattr(asia_stock_kline, "_get_twelve_data_api_key", lambda: "test-only")
+ monkeypatch.setattr(asia_stock_kline, "_twelvedata_daily_limit_active", lambda: False)
+ captured = {}
+
+ def get(_url, **kwargs):
+ captured.update(kwargs["params"])
+ return SimpleNamespace(json=lambda: {"status": "ok", "values": [
+ dict(datetime=f"2025-07-07 {label}", open="10", high="12", low="9", close="11", volume="42")
+ for label in ["09:30:00", "10:00:00"]
+ ]})
+
+ monkeypatch.setattr(asia_stock_kline.requests, "get", get)
+ result = asia_stock_kline.fetch_twelvedata_klines(
+ is_hk=True, tencent_code="HK00700", timeframe="1H", limit=10,
+ before_time=int(utc("2025-07-07T02:30:00").timestamp()),
+ )
+ assert captured["interval"] == "30min"
+ assert captured["end_date"] == "2025-07-07 10:30:00"
+ assert result == [dict(bars(["2025-07-07T01:30:00"])[0], volume=84)]
+
+
+def test_us_weekly_resample_uses_holiday_week_not_five_rows():
+ from app.services.market_schedule import resample_equity_klines
+ rows = bars(["2025-06-30T00:00:00", "2025-07-01T00:00:00", "2025-07-02T00:00:00", "2025-07-03T00:00:00", "2025-07-07T00:00:00"])
+ result = resample_equity_klines(rows, "USStock", 86400, 604800)
+ assert len(result) == 1
+ assert result[0]["time"] == int(utc("2025-06-30T04:00:00").timestamp())
+ assert result[0]["volume"] == 168
+ assert resample_equity_klines(rows[1:], "USStock", 86400, 604800) == []
diff --git a/backend_api_python/tests/test_event_protocol.py b/backend_api_python/tests/test_event_protocol.py
index 72d7786af..28058671a 100644
--- a/backend_api_python/tests/test_event_protocol.py
+++ b/backend_api_python/tests/test_event_protocol.py
@@ -70,6 +70,30 @@ def test_stock_bar_partition_keeps_same_equity_isolated_by_provider():
assert ":alpaca:" in alpaca.partition_key
+def test_exchange_stock_perpetual_uses_continuous_crypto_clock():
+ stream = BarStreamKey.from_member(
+ {
+ "market": "Crypto",
+ "underlying_market": "USStock",
+ "api_family": "swap",
+ "product_type": "stock_perpetual",
+ "market_type": "swap",
+ "instrument_id": "AAPLUSDT",
+ "symbol": "AAPL/USDT",
+ "exchange_id": "gate",
+ },
+ "1d",
+ )
+
+ assert stream.market == "Crypto"
+ assert supports_bar_close_events("1d", [{
+ "market": "Crypto",
+ "underlying_market": "USStock",
+ "api_family": "swap",
+ "product_type": "stock_perpetual",
+ }])
+
+
def test_order_events_partition_by_execution_account():
intent = OrderIntentEventV1.create(
exchange_id="binance",
diff --git a/backend_api_python/tests/test_gate_stock_runtime_loop.py b/backend_api_python/tests/test_gate_stock_runtime_loop.py
index 2e4bc5c27..5523414db 100644
--- a/backend_api_python/tests/test_gate_stock_runtime_loop.py
+++ b/backend_api_python/tests/test_gate_stock_runtime_loop.py
@@ -1,5 +1,6 @@
from types import SimpleNamespace
from unittest.mock import Mock
+from datetime import datetime, timezone
import time
import pandas as pd
@@ -30,7 +31,27 @@ def handle_data(context, data):
@pytest.mark.parametrize("account_options", [{}, {"environment": "testnet"}, {"enable_demo_trading": True}])
-def test_gate_stock_complete_live_loop_uses_exchange_quotes(monkeypatch, account_options):
+@pytest.mark.parametrize("data_mode", ["current", "stale", "recover"])
+def test_gate_stock_complete_live_loop_uses_exchange_quotes(monkeypatch, account_options, data_mode):
+ from app.services.strategy_runtime import candle_readiness
+
+ class Clock(datetime):
+ @classmethod
+ def now(cls, tz=None):
+ return cls(2026, 9, 14, 15, 0, 5, tzinfo=timezone.utc)
+
+ monkeypatch.setattr(candle_readiness, "datetime", Clock)
+ monkeypatch.setattr(runtime, "datetime", Clock)
+ sessions = []
+ session_type = runtime.StrategyV2LiveSession
+
+ def create_session(**kwargs):
+ session = session_type(**kwargs)
+ session.process = Mock(wraps=session.process)
+ sessions.append(session)
+ return session
+
+ monkeypatch.setattr(runtime, "StrategyV2LiveSession", create_session)
product = {
"symbol": "NVDA/USD", "exchange_id": "gate", "market_type": "spot",
"instrument_id": "NVDA", "product_type": "direct_equity", "api_family": "stock",
@@ -62,17 +83,27 @@ def test_gate_stock_complete_live_loop_uses_exchange_quotes(monkeypatch, account
store.load.return_value = {}
monkeypatch.setattr(runtime, "RuntimeStateStore", lambda **kw: store)
monkeypatch.setattr(runtime, "persist_cancellations", lambda *a: None)
- monkeypatch.setattr(runtime, "time", SimpleNamespace(monotonic=time.monotonic, sleep=lambda _: None, time=time.time))
+ monotonic = [time.monotonic()]
+ monkeypatch.setattr(runtime, "time", SimpleNamespace(monotonic=lambda: monotonic[0], sleep=lambda _: None, time=time.time))
frame = pd.DataFrame(
{"open": 200., "high": 201., "low": 199., "close": 200., "volume": 1.},
index=pd.date_range("2026-09-14 14:00", periods=60, freq="min", tz="UTC"),
)
- monkeypatch.setattr(runtime.StrategyV2BacktestService, "fetch_frequency_frames", lambda *a, **kw: ({"1m": {SYMBOL: frame}}, []))
+ stale_frame = frame.copy()
+ stale_frame.index -= pd.Timedelta(minutes=1)
+ fetches = []
+
+ def fetch_frames(*args, **kwargs):
+ fetches.append(1)
+ ready = data_mode == "current" or (data_mode == "recover" and len(fetches) >= 3)
+ return {"1m": {SYMBOL: frame if ready else stale_frame}}, []
+
+ monkeypatch.setattr(runtime.StrategyV2BacktestService, "fetch_frequency_frames", fetch_frames)
request = Mock(return_value=(200, {"data": {"bids": [{"p": "210.1"}], "asks": [{"p": "210.3"}]}}, ""))
monkeypatch.setattr(GateStockClient, "_request", request)
for _delay in executor._run_strategy_loop(1):
- pass
+ monotonic[0] += 10
if account_options:
assert executor._last_exit_reason == {1: "strategyV2.gateStockTestnetUnsupported"}
@@ -84,7 +115,21 @@ def test_gate_stock_complete_live_loop_uses_exchange_quotes(monkeypatch, account
assert executor._last_exit_reason == {}, logs
assert len(heartbeats) == 4, logs
assert all(prices.get(SYMBOL) == 210.2 for prices, _ in heartbeats[1:]), logs
- assert all(meta["status"] == "healthy" for _, meta in heartbeats[1:]), logs
+ if data_mode == "current":
+ assert all(meta["status"] == "healthy" for _, meta in heartbeats[1:]), logs
+ assert sessions[0].process.call_count == 1
+ elif data_mode == "stale":
+ assert all(meta["status"] == "degraded" for _, meta in heartbeats[1:]), logs
+ assert all(meta["last_error"] == "marketData.incomplete_market_data" for _, meta in heartbeats[1:])
+ assert sessions[0].last_processed is None
+ sessions[0].process.assert_not_called()
+ submit.assert_not_called()
+ assert len(fetches) >= 3
+ else:
+ assert heartbeats[1][1]["status"] == "degraded"
+ assert heartbeats[-1][1]["status"] == "healthy"
+ assert sessions[0].process.call_count == 1
+ assert sessions[0].last_processed == frame.index[-1]
assert request.call_count == 1
diff --git a/backend_api_python/tests/test_strategy_bar_recovery.py b/backend_api_python/tests/test_strategy_bar_recovery.py
new file mode 100644
index 000000000..7ab3dcfec
--- /dev/null
+++ b/backend_api_python/tests/test_strategy_bar_recovery.py
@@ -0,0 +1,191 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+import pytest
+
+from app.events.bar_clock import BarStreamKey
+from app.events.bar_recovery import BarRecoveryRoute, StrategyBarRecoveryPublisher
+from app.events.protocol import MarketBarClosedV1
+
+
+class FakeRepository:
+ def __init__(self, routes):
+ self.routes = list(routes)
+
+ def active_bar_routes(self):
+ return list(self.routes)
+
+
+class FakeProducer:
+ def __init__(self, results=None):
+ self.results = list(results or [True])
+ self.batches = []
+
+ def publish_batch(self, events, *, timeout):
+ self.batches.append((list(events), timeout))
+ return self.results.pop(0) if self.results else True
+
+ def close(self):
+ return None
+
+
+def _publisher(partition_key, *, producer=None):
+ return StrategyBarRecoveryPublisher(
+ repository=FakeRepository([
+ {"partition_key": partition_key, "timeframe": "1m"},
+ ]),
+ producer=producer or FakeProducer(),
+ grace_seconds=0,
+ )
+
+
+@pytest.mark.parametrize("venue", ["binance", "bitget", "bybit", "okx", "gate", "htx"])
+def test_recovery_publishes_same_contract_for_every_crypto_venue(venue):
+ partition_key = MarketBarClosedV1.partition_key(
+ venue=venue,
+ market="Crypto",
+ market_type="swap",
+ instrument_id="BTCUSDT",
+ timeframe="1m",
+ )
+ producer = FakeProducer()
+ recovery = _publisher(partition_key, producer=producer)
+
+ assert recovery.publish_due(datetime(2026, 9, 29, 12, 1, 6, tzinfo=timezone.utc)) == 1
+
+ event = producer.batches[0][0][0]
+ assert event.partition_key == partition_key
+ assert event.payload["venue"] == venue
+ assert event.producer == "strategy-bar-recovery"
+
+
+def test_recovery_parser_preserves_colons_inside_canonical_instrument():
+ partition_key = "bar:crypto:crypto:swap:crypto:btc/usdt@swap:1m"
+
+ route = BarRecoveryRoute.parse(partition_key, "1m")
+ event = route.event(100)
+
+ assert route.instrument_id == "crypto:btc/usdt@swap"
+ assert event.partition_key == partition_key
+
+
+def test_recovery_retries_same_token_after_delivery_failure():
+ partition_key = "bar:gate:crypto:swap:btc_usdt:1m"
+ producer = FakeProducer([False, True])
+ recovery = _publisher(partition_key, producer=producer)
+ now = datetime(2026, 9, 29, 12, 1, 6, tzinfo=timezone.utc)
+
+ assert recovery.publish_due(now) == 0
+ assert recovery.publish_due(now) == 1
+ assert len(producer.batches) == 2
+ assert producer.batches[0][0][0].event_id == producer.batches[1][0][0].event_id
+
+
+def test_recovery_catches_up_each_missed_boundary_after_transport_outage():
+ partition_key = "bar:gate:crypto:swap:btc_usdt:1m"
+ producer = FakeProducer([False, True])
+ recovery = _publisher(partition_key, producer=producer)
+ first = datetime(2026, 9, 29, 12, 1, 6, tzinfo=timezone.utc)
+ restored = datetime(2026, 9, 29, 12, 3, 6, tzinfo=timezone.utc)
+
+ assert recovery.publish_due(first) == 0
+ assert recovery.publish_due(restored) == 3
+
+ tokens = [event.payload["closed_bar_token"] for event in producer.batches[1][0]]
+ assert tokens == list(range(tokens[0], tokens[0] + 3))
+
+
+def test_primary_bar_event_suppresses_recovery_duplicate():
+ event = MarketBarClosedV1.create(
+ venue="okx",
+ market="Crypto",
+ market_type="swap",
+ instrument_id="BTCUSDT",
+ symbol="BTC/USDT",
+ timeframe="1m",
+ closed_bar_token=100,
+ closed_at=datetime.fromtimestamp(101 * 60, tz=timezone.utc),
+ )
+ producer = FakeProducer()
+ recovery = _publisher(event.partition_key, producer=producer)
+ recovery.observe(event)
+
+ assert recovery.publish_due(datetime.fromtimestamp(101 * 60 + 5, tz=timezone.utc)) == 0
+ assert producer.batches == []
+
+
+@pytest.mark.parametrize(
+ ("market", "outside_session", "inside_session"),
+ [
+ (
+ "USStock",
+ datetime(2026, 9, 29, 12, 1, 6, tzinfo=timezone.utc),
+ datetime(2026, 9, 29, 13, 31, 6, tzinfo=timezone.utc),
+ ),
+ (
+ "HKStock",
+ datetime(2026, 9, 29, 4, 31, 6, tzinfo=timezone.utc),
+ datetime(2026, 9, 29, 5, 1, 6, tzinfo=timezone.utc),
+ ),
+ ],
+)
+def test_direct_equity_recovery_obeys_market_sessions(
+ market,
+ outside_session,
+ inside_session,
+):
+ stream = BarStreamKey.from_member(
+ {
+ "market": "Crypto",
+ "underlying_market": market,
+ "api_family": "stock",
+ "product_type": "direct_equity",
+ "market_type": "spot",
+ "instrument_id": "AAPL" if market == "USStock" else "0700",
+ "symbol": "AAPL/USD" if market == "USStock" else "0700/HKD",
+ "exchange_id": "gate",
+ },
+ "1m",
+ )
+
+ outside = _publisher(stream.partition_key, producer=FakeProducer())
+ inside_producer = FakeProducer()
+ inside = _publisher(stream.partition_key, producer=inside_producer)
+
+ assert outside.publish_due(outside_session) == 0
+ assert inside.publish_due(inside_session) == 1
+ assert inside_producer.batches[0][0][0].payload["market"] == market
+
+
+@pytest.mark.parametrize("product_type,api_family,market_type", [
+ ("stock_perpetual", "swap", "swap"),
+ ("tokenized_equity", "spot", "spot"),
+ ("tokenized_equity", "reality", "spot"),
+])
+def test_exchange_equity_products_use_continuous_crypto_boundaries(
+ product_type,
+ api_family,
+ market_type,
+):
+ stream = BarStreamKey.from_member(
+ {
+ "market": "Crypto",
+ "underlying_market": "USStock",
+ "api_family": api_family,
+ "product_type": product_type,
+ "market_type": market_type,
+ "instrument_id": "AAPLUSDT",
+ "symbol": "AAPL/USDT",
+ "exchange_id": "gate",
+ },
+ "1m",
+ )
+ producer = FakeProducer()
+ recovery = _publisher(stream.partition_key, producer=producer)
+
+ assert stream.market == "Crypto"
+ assert recovery.publish_due(
+ datetime(2026, 10, 3, 12, 1, 6, tzinfo=timezone.utc)
+ ) == 1
+ assert producer.batches[0][0][0].payload["market"] == "Crypto"
diff --git a/backend_api_python/tests/test_strategy_documentation_contract.py b/backend_api_python/tests/test_strategy_documentation_contract.py
new file mode 100644
index 000000000..089fdc708
--- /dev/null
+++ b/backend_api_python/tests/test_strategy_documentation_contract.py
@@ -0,0 +1,95 @@
+import ast
+from pathlib import Path
+
+import pytest
+
+
+BACKEND_DIR = Path(__file__).parents[1]
+DOCS_DIR = Path(__file__).parents[2] / "docs" / "trading"
+RUNTIME_PATH = BACKEND_DIR / "app" / "services" / "strategy_v2" / "runtime.py"
+CONTRACT_PATH = BACKEND_DIR / "app" / "services" / "strategy_v2" / "contract.py"
+
+
+def _runtime_order_option_names():
+ tree = ast.parse(RUNTIME_PATH.read_text(encoding="utf-8"))
+ for node in tree.body:
+ if not isinstance(node, ast.Assign):
+ continue
+ matches = any(
+ isinstance(target, ast.Name) and target.id == "_ORDER_OPTION_NAMES"
+ for target in node.targets
+ )
+ if not matches:
+ continue
+ call = node.value
+ assert isinstance(call, ast.Call)
+ assert call.args
+ return set(ast.literal_eval(call.args[0]))
+ raise AssertionError("_ORDER_OPTION_NAMES was not found in Strategy API V2 runtime")
+
+
+def _literal_name_set(path, assignment_name):
+ tree = ast.parse(path.read_text(encoding="utf-8"))
+ for node in tree.body:
+ if not isinstance(node, ast.Assign):
+ continue
+ matches = any(
+ isinstance(target, ast.Name) and target.id == assignment_name
+ for target in node.targets
+ )
+ if not matches:
+ continue
+ return set(ast.literal_eval(node.value))
+ raise AssertionError(f"{assignment_name} was not found in {path.name}")
+
+
+def _discovery_context_api_names():
+ tree = ast.parse(CONTRACT_PATH.read_text(encoding="utf-8"))
+ for node in tree.body:
+ if isinstance(node, ast.ClassDef) and node.name == "DiscoveryContext":
+ return {
+ item.name
+ for item in node.body
+ if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef))
+ and not item.name.startswith("_")
+ }
+ raise AssertionError("DiscoveryContext was not found in Strategy API V2 contract")
+
+
+@pytest.mark.parametrize(
+ "filename",
+ ["STRATEGY_DEV_GUIDE.md", "STRATEGY_DEV_GUIDE_CN.md"],
+)
+def test_strategy_guide_documents_every_order_option(filename):
+ guide = (DOCS_DIR / filename).read_text(encoding="utf-8")
+
+ for option_name in sorted(_runtime_order_option_names()):
+ assert f"{option_name}" in guide
+
+
+@pytest.mark.parametrize(
+ "filename",
+ ["STRATEGY_DEV_GUIDE.md", "STRATEGY_DEV_GUIDE_CN.md"],
+)
+def test_strategy_guide_explains_protection_units_and_non_executable_params(filename):
+ guide = (DOCS_DIR / filename).read_text(encoding="utf-8")
+
+ assert "0..1" in guide
+ assert "0..5" in guide
+ assert "stop_loss" in guide
+ assert "# @param stop_loss_pct" in guide
+ assert "set_default_protection()" in guide
+
+
+@pytest.mark.parametrize(
+ "filename",
+ ["STRATEGY_DEV_GUIDE.md", "STRATEGY_DEV_GUIDE_CN.md"],
+)
+def test_strategy_guide_names_every_runtime_global(filename):
+ guide = (DOCS_DIR / filename).read_text(encoding="utf-8")
+
+ for api_name in sorted(_literal_name_set(CONTRACT_PATH, "_RUNTIME_GLOBAL_CALL_NAMES")):
+ assert api_name in guide
+
+ for api_name in sorted(_discovery_context_api_names()):
+ assert api_name in guide
diff --git a/backend_api_python/tests/test_strategy_v2_contract.py b/backend_api_python/tests/test_strategy_v2_contract.py
index 39cc203bc..ec759f7d9 100644
--- a/backend_api_python/tests/test_strategy_v2_contract.py
+++ b/backend_api_python/tests/test_strategy_v2_contract.py
@@ -178,6 +178,23 @@ def handle_data(context, data):
compile_strategy_v2(code)
+def test_contract_rejects_unsupported_order_protection_keyword():
+ code = '''
+def initialize(context):
+ context.set_universe(["Crypto:BTC/USDT@swap"])
+ context.subscribe(frequency="1m")
+
+def handle_data(context, data):
+ order_target_percent("Crypto:BTC/USDT@swap", 0.1, stop_loss=75000)
+'''
+
+ with pytest.raises(
+ StrategyV2ContractError,
+ match="strategyV2.apiCallInvalid:order_target_percent:unsupportedArgument:stop_loss",
+ ):
+ compile_strategy_v2(code)
+
+
def test_contract_rejects_symbol_index_on_single_history_dataframe():
code = '''
def initialize(context):
diff --git a/backend_api_python/tests/test_strategy_v2_runtime.py b/backend_api_python/tests/test_strategy_v2_runtime.py
index 143904ae6..066973d91 100644
--- a/backend_api_python/tests/test_strategy_v2_runtime.py
+++ b/backend_api_python/tests/test_strategy_v2_runtime.py
@@ -4,7 +4,11 @@
import pytest
from app.services.instrument_rules import InstrumentRules
-from app.services.strategy_v2 import StrategyV2BacktestRunner, StrategyV2LiveSession
+from app.services.strategy_v2 import (
+ StrategyV2BacktestRunner,
+ StrategyV2ContractError,
+ StrategyV2LiveSession,
+)
from app.services.strategy_v2.data import MultiAssetDataPortal
from app.services.strategy_v2.models import ScheduleSpec
from app.services.strategy_v2.runtime import MultiAssetSimulationBroker, OrderIntent, Position
@@ -109,6 +113,82 @@ def test_data_portal_caches_timestamps_and_slices_point_in_time_history():
assert list(current.index) == list(cached_timestamps[499:501])
+def test_legacy_swap_position_lookup_returns_the_only_open_directional_leg():
+ symbol = "Crypto:BTC/USDT@swap"
+ code = f'''
+def initialize(context):
+ context.set_universe(["{symbol}"])
+ context.subscribe(frequency="1m")
+ context.set_metadata(direction_mode="long_only")
+
+def handle_data(context, data):
+ pass
+'''
+ session = StrategyV2LiveSession(
+ code=code,
+ frames={symbol: _frame([100, 101])},
+ initial_capital=1000,
+ )
+ session.synchronize_positions({
+ f"{symbol}::long": {
+ "amount": 2,
+ "position_side": "long",
+ "avg_cost": 100,
+ "last_price": 101,
+ },
+ })
+
+ position = session.context.get_position(symbol)
+
+ assert position.amount == 2
+ assert position.position_side == "long"
+
+
+def test_legacy_swap_position_lookup_requires_side_when_both_legs_are_open():
+ symbol = "Crypto:BTC/USDT@swap"
+ code = f'''
+def initialize(context):
+ context.set_universe(["{symbol}"])
+ context.subscribe(frequency="1m")
+ context.set_metadata(direction_mode="both")
+
+def handle_data(context, data):
+ pass
+'''
+ session = StrategyV2LiveSession(
+ code=code,
+ frames={symbol: _frame([100, 101])},
+ initial_capital=1000,
+ )
+ session.synchronize_positions({
+ f"{symbol}::long": {"amount": 2, "position_side": "long"},
+ f"{symbol}::short": {"amount": 1, "position_side": "short"},
+ })
+
+ with pytest.raises(StrategyV2ContractError, match="strategyV2.positionSideRequired"):
+ session.context.get_position(symbol)
+
+
+def test_runtime_rejects_dynamic_unsupported_order_keyword():
+ symbol = "Crypto:BTC/USDT@swap"
+ code = f'''
+def initialize(context):
+ context.set_universe(["{symbol}"])
+ context.subscribe(frequency="1m")
+
+def handle_data(context, data):
+ pass
+'''
+ session = StrategyV2LiveSession(
+ code=code,
+ frames={symbol: _frame([100, 101])},
+ initial_capital=1000,
+ )
+
+ with pytest.raises(StrategyV2ContractError, match="unsupportedArgument:stop_loss"):
+ session.context.order_target_percent(symbol, 0.1, **{"stop_loss": 95})
+
+
def test_multi_asset_strategy_controls_symbols_and_rebalances_without_ui_market_fields():
code = """
def initialize(context):
diff --git a/backend_api_python/tests/test_us_daily_execution.py b/backend_api_python/tests/test_us_daily_execution.py
index 42812d398..67960f1f3 100644
--- a/backend_api_python/tests/test_us_daily_execution.py
+++ b/backend_api_python/tests/test_us_daily_execution.py
@@ -86,7 +86,7 @@ def test_all_constituents_must_have_previous_session_data():
def test_daily_loader_includes_yesterdays_session_and_excludes_unfinished_today(monkeypatch):
market_data.clear_shared_strategy_frame_cache()
- monkeypatch.setattr(market_data, "_market_bar_cutoff", lambda *_: equity_daily_bar_cutoff("USStock", utc("2026-09-18T14:00")))
+ monkeypatch.setattr(market_data, "_market_bar_cutoff", lambda *_, **kwargs: equity_daily_bar_cutoff("USStock", utc("2026-09-18T14:00")))
monkeypatch.setattr(market_data._cache, "get", lambda *_: None)
monkeypatch.setattr(market_data._cache, "put", lambda *_: None)
times = ["2026-09-16T13:30", "2026-09-17T13:30", "2026-09-18T13:30"]
@@ -149,7 +149,7 @@ def execute(**kwargs):
def test_delayed_daily_provider_is_retried_in_shared_cache(monkeypatch):
market_data.clear_shared_strategy_frame_cache()
- monkeypatch.setattr(market_data, "_market_bar_cutoff", lambda *_: equity_daily_bar_cutoff("USStock", utc("2026-09-18T14:00")))
+ monkeypatch.setattr(market_data, "_market_bar_cutoff", lambda *_, **kwargs: equity_daily_bar_cutoff("USStock", utc("2026-09-18T14:00")))
calls = []
def fetch(*args, **kwargs):
@@ -237,7 +237,7 @@ def test_token_equity_candles_are_not_replaced_with_underlying_stock_calendar(fa
def test_hk_loader_excludes_intraday_candle_labelled_at_local_midnight(monkeypatch):
market_data.clear_shared_strategy_frame_cache()
- monkeypatch.setattr(market_data, "_market_bar_cutoff", lambda *_: equity_daily_bar_cutoff("HKStock", utc("2026-09-18T01:35")))
+ monkeypatch.setattr(market_data, "_market_bar_cutoff", lambda *_, **kwargs: equity_daily_bar_cutoff("HKStock", utc("2026-09-18T01:35")))
monkeypatch.setattr(market_data._cache, "get", lambda *_: None)
monkeypatch.setattr(market_data._cache, "put", lambda *_: None)
times = ["2026-09-16T16:00", "2026-09-17T16:00"]
diff --git a/docs/trading/STRATEGY_DEV_GUIDE.md b/docs/trading/STRATEGY_DEV_GUIDE.md
index 8496a1b37..dee56208e 100644
--- a/docs/trading/STRATEGY_DEV_GUIDE.md
+++ b/docs/trading/STRATEGY_DEV_GUIDE.md
@@ -393,6 +393,9 @@ Live sessions process each closed bar once and preserve g state whi
- Real-time prices are reserved for stop loss, take profit, trailing protection, and equity risk. They must not make an unfinished candle look complete or revise a confirmed strategy signal.
- Multi-timeframe strategies apply the completed-bar rule separately to every timeframe. A lower-timeframe advance never exposes a still-forming four-hour or daily bar.
- The runtime filters unfinished candles centrally. Strategy source must not drop the final row, guess closure from local wall-clock time, or write a live price into OHLC; those workarounds break parity across backtests, restart replay, and live execution.
+- Candle boundaries follow the selected venue, including its daily timezone and weekly anchor. A close event wakes the runtime; it does not prove that the provider has published the final candle. The runtime checks the latest required candle for every subscribed timeframe, respects explicit provider confirmation, and retries missing data without advancing signal progress. Normal quotes alone do not clear a candle-data fault.
+- US and Hong Kong equities use exchange sessions, holidays, early closes, daylight saving time, and the Hong Kong lunch break. Daily and weekly bars retain the 15-minute post-close data allowance. Aggregated bars must contain all expected constituent bars in their session; they must not bridge a missing bar or use five daily rows as a calendar week. Exchange-listed stock perpetuals and tokenized equities use their venue candle calendar, not the underlying stock-market session calendar.
+- Recovery does not backfill historical real orders. Confirmed-bar deduplication remains in effect, and old cached candle data is not reused across candle-contract revisions.
- A future tick-driven strategy product needs a separate API, replay model, and contract. Do not simulate it inside ordinary Strategy API V2 source.
---
@@ -414,6 +417,8 @@ Common context fields:
Use data.current(symbol, field, frequency="1h") for a current visible value, data.history(symbols, count, fields, frequency="4h") for a selected timeframe, and data[symbol] for the driving timeframe's current visible DataFrame.
+Use is_trade() to filter scheduled callbacks. It returns True when the current event timestamp exists in at least one subscribed data frame and False otherwise. It answers “does this event have tradable data?”, not whether an exchange is open, an account may submit orders, or a product is suspended; live eligibility is still validated by deployment and execution layers.
+
Persist state across callbacks on g:
~~~python
@@ -598,6 +603,15 @@ Common order options:
| execution_algo | market, limit, or maker_then_market |
| maker_wait_sec | maker wait before market fallback |
| maker_offset_bps | maker-price offset in basis points |
+| protection | protection mapping containing the stop, take-profit, trailing, and time fields below |
+| stop_loss_pct | fixed stop-loss decimal ratio from position cost |
+| take_profit_pct | fixed take-profit decimal ratio from position cost |
+| trailing_stop_pct | trailing distance from the subsequent position peak/trough |
+| trailing_activation_pct | profit threshold from position cost that activates trailing |
+| time_limit_seconds | maximum holding time in seconds |
+| trailing_rebase_on_scale_in | whether a same-side scale-in resets trailing state and the holding clock; defaults to True |
+
+type and price are compatibility aliases for order_type and the limit-order limit_price, respectively. New strategies should use the canonical names. The compiler and runtime reject keywords outside this table instead of silently ignoring them.
Example with a stable client reference:
@@ -635,7 +649,7 @@ In live execution, an active same-leg request suppresses duplicate requests unti
## 13. Stop, take-profit, trailing, and time protection
-Attach protection to an entry:
+Protection arguments are accepted by all five order helpers, but should be declared on an entry or same-side scale-in. The most direct form uses inline order keywords:
~~~python
order_target_percent(
@@ -650,6 +664,22 @@ order_target_percent(
)
~~~
+A literal mapping is also supported. Inline fields on the same call override matching values in that mapping:
+
+~~~python
+order_target_percent(
+ g.symbol,
+ 0.8,
+ reason="breakout_long_entry",
+ protection={
+ "stop_loss_pct": 0.03,
+ "take_profit_pct": 0.08,
+ },
+ trailing_stop_pct=0.025,
+ trailing_activation_pct=0.02,
+)
+~~~
+
Or set defaults inside an executable handler/scheduled callback before later entries (do not call this in initialize; manifest discovery does not retain runtime protection state):
~~~python
@@ -659,7 +689,38 @@ set_default_protection(
)
~~~
-Percentage fields are ratios: 0.03 means 3%. Values are clamped to safe ranges, and negatives become zero.
+set_default_protection applies to later orders that have no effective per-order protection; per-order protection takes precedence. Call set_default_protection() to clear the defaults. If a Crypto swap strategy requests protection, put effective protection directly on every opening leg. AI validation does not treat defaults alone as proof that each swap entry leg is protected.
+
+Parameter contract:
+
+| Parameter | Unit and range | Behavior |
+| --- | --- | --- |
+| stop_loss_pct | decimal ratio, 0..1 | long trigger is cost × (1-pct); short trigger is cost × (1+pct) |
+| take_profit_pct | decimal ratio, 0..5 | long trigger is cost × (1+pct); short trigger is cost × (1-pct) |
+| trailing_stop_pct | decimal ratio, 0..1 | after activation, follows the highest price for longs and the lowest price for shorts |
+| trailing_activation_pct | decimal ratio, 0..5 | favorable move from cost required to activate trailing; activation without a trailing distance creates no protection |
+| time_limit_seconds | non-negative integer seconds | exits after the holding limit; 0 disables it |
+| trailing_rebase_on_scale_in | boolean, default True | True resets trailing extrema, activation, and the holding clock on a same-side scale-in; False preserves an active trail and the original clock |
+
+All pct values are decimal ratios: 0.03 means 3%, not 0.03% or 3. A value of 0 disables that rule. The runtime clamps negative and oversized values to the ranges above, but source and AI-generated strategies should provide valid values rather than rely on correction.
+
+Names must be exact. Strategy API V2 does not accept stop_loss, take_profit, stop_price, or take_profit_price as order arguments; they fail with strategyV2.apiCallInvalid:...:unsupportedArgument:.... Fixed absolute-price protection is not currently part of the Strategy API V2 source contract. Use pct protection or implement an explicit strategy-signal exit.
+
+# @param stop_loss_pct ... only declares a run-panel parameter; it does not enable a stop automatically. A handler must read it and pass it to the entry order:
+
+~~~python
+def handle_data(context, data):
+ stop_loss_pct = float(context.params.get("stop_loss_pct", 0.03))
+ if should_enter:
+ order_target_percent(
+ g.symbol,
+ 0.5,
+ reason="protected_entry",
+ stop_loss_pct=stop_loss_pct,
+ )
+~~~
+
+Fixed stop/take-profit protection, trailing protection, and signal exits are distinct mechanisms. Only effective order fields or runtime defaults create protection state; a parameter declaration, comment, or ordinary exit condition does not prove that fixed protection is attached. After a protection exit, consume_last_exit_reason(symbol) returns and clears stop_loss, take_profit, trailing_stop, or time_limit.
Backtest behavior:
diff --git a/docs/trading/STRATEGY_DEV_GUIDE_CN.md b/docs/trading/STRATEGY_DEV_GUIDE_CN.md
index 3d837a169..b7494301d 100644
--- a/docs/trading/STRATEGY_DEV_GUIDE_CN.md
+++ b/docs/trading/STRATEGY_DEV_GUIDE_CN.md
@@ -395,6 +395,9 @@ def initialize(context):
- 实时价格仅供止损、止盈、追踪止损和权益风控使用；它不能把一根尚未收盘的 K 线伪装成完成 bar，也不能改变已经确认的策略信号。
- 多周期策略对每个周期分别执行已完成 K 线约束；低周期推进不会提前暴露仍在形成中的 4 小时线或日线。
- 运行时会统一过滤未收盘 K 线。策略源码不要自行删除最后一行、用本机时间猜测收盘，或把实时价格写入 OHLC；这些补丁会破坏回测、重启重放和实盘的一致性。
+- K 线边界遵循所选交易所的日线时区和周线起点。闭合事件仅用于唤醒运行时，不代表行情源已经发布最终 K 线；运行时会逐一检查所有订阅周期的最新必需 K 线，尊重行情源显式闭合确认，缺失时重试而不推进信号处理进度。仅有报价正常不能清除 K 线数据故障。
+- 美股和港股遵循交易日历、节假日、提前收市、夏令时及港股午休；日线和周线保留收市后 15 分钟的数据等待时间。聚合 K 线必须包含该交易时段内全部应有的子 K 线，不能跨缺口拼接，也不能把固定 5 根日线当作自然周。加密交易所的股票永续合约和代币化股票遵循所在交易所的 K 线日历，不套用标的股票市场的休市日历。
+- 行情恢复不会补发历史实盘订单；已确认 K 线的去重仍然有效，K 线契约版本变更后不复用旧版行情缓存。
- 如果产品以后提供独立的逐 tick 策略契约，应使用单独的 API、回测模型和文档；不要在普通 Strategy API V2 源码中自行模拟。
---
@@ -416,6 +419,8 @@ def initialize(context):
data.current(symbol, field, frequency="1h") 读取当前可见值；data.history(symbols, count, fields, frequency="4h") 读取指定周期历史；data[symbol] 返回驱动周期的当前可见 DataFrame。
+is_trade() 用于定时回调中的事件过滤：当前事件时间戳只要存在于至少一个已订阅数据帧就返回 True，否则返回 False。它判断的是“当前事件是否有可交易数据”，不是交易所是否开市、账户是否可下单或产品是否暂停；这些实盘资格仍由部署和执行层校验。
+
跨回调状态放在 g：
~~~python
@@ -600,6 +605,15 @@ order_target_percent(
| execution_algo | market、limit 或 maker_then_market |
| maker_wait_sec | maker 等待多久后回退为市价 |
| maker_offset_bps | maker 价格偏移，单位为基点 |
+| protection | 保护参数映射；可包含本节列出的止损、止盈、追踪和时间字段 |
+| stop_loss_pct | 相对持仓成本的固定止损小数比率 |
+| take_profit_pct | 相对持仓成本的固定止盈小数比率 |
+| trailing_stop_pct | 相对持仓后续峰值/谷值的追踪距离 |
+| trailing_activation_pct | 相对持仓成本的追踪激活盈利阈值 |
+| time_limit_seconds | 持仓最长保留秒数 |
+| trailing_rebase_on_scale_in | 同方向加仓时是否重置追踪状态和持仓计时，默认 True |
+
+type 和 price 是分别对应 order_type 和限价 limit_price 的兼容别名；新策略应使用规范名称。编译器和运行时都会拒绝表外关键字，而不是静默忽略。
使用稳定客户端引用的示例：
@@ -637,7 +651,7 @@ def monitor_entry(cancel_requested):
## 13. 止损、止盈、追踪和时间保护
-随开仓声明：
+保护参数可以传给全部五个订单函数，但应随开仓或同方向加仓声明。最直接的写法是把字段作为订单关键字传入：
~~~python
order_target_percent(
@@ -652,6 +666,22 @@ order_target_percent(
)
~~~
+也可以使用一个字面量映射；同一次调用中的内联字段会覆盖映射中的同名字段：
+
+~~~python
+order_target_percent(
+ g.symbol,
+ 0.8,
+ reason="breakout_long_entry",
+ protection={
+ "stop_loss_pct": 0.03,
+ "take_profit_pct": 0.08,
+ },
+ trailing_stop_pct=0.025,
+ trailing_activation_pct=0.02,
+)
+~~~
+
或在可执行处理器/定时回调中、下单之前设置后续开仓的默认保护（不要在 initialize 中调用；清单发现阶段不会保留运行时保护状态）：
~~~python
@@ -661,7 +691,38 @@ set_default_protection(
)
~~~
-所有 pct 都使用小数比率，0.03 表示 3%。保护值会限制在安全范围内；负值按 0 处理。
+set_default_protection 会作用于之后没有有效逐单保护的订单；逐单保护优先。需要清除默认值时调用 set_default_protection()。Crypto swap 策略如果声明需要保护，建议在每个开仓腿的订单上直接传入有效保护；AI 验证不会只凭默认保护推断每条开仓腿都已受保护。
+
+参数契约：
+
+| 参数 | 单位和范围 | 行为 |
+| --- | --- | --- |
+| stop_loss_pct | 小数比率，0..1 | 多头触发价为成本价 × (1-pct)，空头为成本价 × (1+pct) |
+| take_profit_pct | 小数比率，0..5 | 多头触发价为成本价 × (1+pct)，空头为成本价 × (1-pct) |
+| trailing_stop_pct | 小数比率，0..1 | 激活后，多头跟随最高价、空头跟随最低价 |
+| trailing_activation_pct | 小数比率，0..5 | 从成本价开始计算的盈利激活阈值；只有它而没有追踪距离不会产生保护 |
+| time_limit_seconds | 非负整数秒 | 达到持仓时限后退出；0 表示关闭 |
+| trailing_rebase_on_scale_in | 布尔值，默认 True | True 在同方向加仓时重置追踪峰谷、激活状态和持仓计时；False 保留已经激活的追踪状态和原计时 |
+
+所有 pct 都是小数比率：0.03 表示 3%，不是 0.03% 或 3。0 会关闭相应规则。运行时会把负值归零并把过大值限制到上表范围，但策略源码和 AI 生成结果不应依赖这种纠正；应直接提供范围内的值。
+
+字段名必须精确。Strategy API V2 不接受 stop_loss、take_profit、stop_price 或 take_profit_price 作为订单参数；这些写法会以 strategyV2.apiCallInvalid:...:unsupportedArgument:... 拒绝。固定绝对价格保护目前不是 Strategy API V2 的源码契约，应改用 pct 保护，或在策略信号逻辑中显式退出。
+
+# @param stop_loss_pct ... 只会声明运行面板参数，不会自动启用止损。处理器必须读取它并传给开仓订单：
+
+~~~python
+def handle_data(context, data):
+ stop_loss_pct = float(context.params.get("stop_loss_pct", 0.03))
+ if should_enter:
+ order_target_percent(
+ g.symbol,
+ 0.5,
+ reason="protected_entry",
+ stop_loss_pct=stop_loss_pct,
+ )
+~~~
+
+固定止损/止盈、追踪保护和信号离场是不同机制。只有订单上的有效字段或运行时默认保护会建立保护状态；参数声明、注释或普通离场条件都不能证明已经附加固定保护。保护离场后可用 consume_last_exit_reason(symbol) 读取并清除 stop_loss、take_profit、trailing_stop 或 time_limit 原因。
回测规则：