Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion application/rebalance_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
from application.state_persistence import GcsStateStore, build_gcs_state_store_from_env
from application.strategy_run_persistence import (
build_strategy_run_state,
claim_live_strategy_run,
is_duplicate_live_run,
persist_strategy_run_state,
read_latest_strategy_run_state,
Expand Down Expand Up @@ -512,13 +513,26 @@ def log_message(message: str) -> None:
masked_account = mask_account_id(account)
existing_run = None
if persist_strategy_runs and not settings.dry_run_only:
claim_acquired = claim_live_strategy_run(
store=store,
account=masked_account,
strategy_profile=strategy_runtime.profile,
run_period=run_period,
now=now,
)
existing_run = read_latest_strategy_run_state(
store=store,
account=masked_account,
strategy_profile=strategy_runtime.profile,
run_period=run_period,
)
if is_duplicate_live_run(existing_run):
if not claim_acquired and existing_run is None:
existing_run = {
"stage": "PENDING_SUBMISSION",
"as_of": now.isoformat(),
"claim_only": True,
}
if not claim_acquired or is_duplicate_live_run(existing_run):
duplicate_stage = str(existing_run.get("stage") or "NO_ACTION")
duplicate_skipped_orders = [
{
Expand Down
9 changes: 9 additions & 0 deletions application/state_persistence.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,15 @@ def write_json(self, key: str, payload: dict[str, Any]) -> bool:
raise StatePersistenceError(f"GCS write failed for {key}: {exc}") from exc
return True

def create_json(self, key: str, payload: dict[str, Any]) -> bool:
"""Atomically create a JSON object; return False if it already exists."""
uri = self._object_uri(key)
data = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
try:
return bool(_object_store().create_text(uri, data, content_type="application/json"))
except Exception as exc:
raise StatePersistenceError(f"GCS atomic create failed for {key}: {exc}") from exc


def build_gcs_state_store_from_env(
env: Callable[[str, str | None], str | None] = os.getenv,
Expand Down
31 changes: 31 additions & 0 deletions application/strategy_run_persistence.py
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,37 @@ def strategy_run_history_key(
)


def strategy_run_claim_key(
*, account: str, strategy_profile: str, run_period: str,
) -> str:
"""Permanent live claim key; a failed/unknown submission must remain blocked."""
return (
f"strategy-runs/claims/{safe_key(account)}/{safe_key(strategy_profile)}/"
f"{safe_key(run_period)}.json"
)


def claim_live_strategy_run(
*, store: GcsStateStore, account: str, strategy_profile: str,
run_period: str, now: datetime | None = None,
) -> bool:
"""Acquire the durable pre-order claim using object-store create-if-absent."""
payload = {
"stage": "PENDING_SUBMISSION",
"account": account,
"strategy_profile": strategy_profile,
"run_period": run_period,
"as_of": (now or utcnow()).isoformat(),
"no_order_submitted": True,
}
return store.create_json(
strategy_run_claim_key(
account=account, strategy_profile=strategy_profile, run_period=run_period,
),
payload,
)


def read_latest_strategy_run_state(
*,
store: GcsStateStore,
Expand Down
6 changes: 3 additions & 3 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@ dependencies = [
"pytest",
"pytz",
"requests",
"quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@92458590a463e7219f0369a3505031ee74414135",
"us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@17ddb86c72d44b2c7b78ba7a10d8f71b21180166",
"quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@3acab1923a97b805b077c85c6c19657be0143bac",
"us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@be1a2c9f7c388d4dd78bda6c2dd7ccad0d4b13b4",
]
license = "MIT"
authors = [
Expand Down Expand Up @@ -82,5 +82,5 @@ show_missing = true

[tool.uv]
override-dependencies = [
"quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@92458590a463e7219f0369a3505031ee74414135",
"quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@3acab1923a97b805b077c85c6c19657be0143bac",
]
6 changes: 3 additions & 3 deletions qsl.toml
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@ upgrade_ring = "ring_d"
allow_legacy = false

[qsl.requires]
quant_platform_kit = "92458590a463e7219f0369a3505031ee74414135"
us_equity_strategies = "17ddb86c72d44b2c7b78ba7a10d8f71b21180166"
quant_platform_kit = "3acab1923a97b805b077c85c6c19657be0143bac"
us_equity_strategies = "be1a2c9f7c388d4dd78bda6c2dd7ccad0d4b13b4"

[qsl.compat]
bundle = "2026.07.4"
bundle = "2026.08.0"
2 changes: 1 addition & 1 deletion tests/test_qsl_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ def test_qsl_metadata_has_runtime_platform_fields() -> None:
assert qsl["tier"] == "runtime"
assert qsl["upgrade_ring"] == "ring_d"
assert qsl.get("repo") == "FirstradePlatform"
assert qsl["compat"]["bundle"] == "2026.07.4"
assert qsl["compat"]["bundle"] == "2026.08.0"
requires = qsl["requires"]
assert "quant_platform_kit" in requires
assert "us_equity_strategies" in requires
6 changes: 6 additions & 0 deletions tests/test_rebalance_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,12 @@ def write_json(self, key, payload):
self.writes.append((key, dict(payload)))
return True

def create_json(self, key, payload):
if key in self.payloads:
return False
self.payloads[key] = dict(payload)
return True


def _latest_strategy_run_payloads(store: FakeStateStore) -> list[dict]:
return [payload for key, payload in store.writes if key.endswith("latest.json")]
Expand Down
33 changes: 33 additions & 0 deletions tests/test_strategy_run_claim.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
from application.strategy_run_persistence import (
claim_live_strategy_run,
strategy_run_claim_key,
)


class AtomicFakeStore:
def __init__(self):
self.payloads = {}

def create_json(self, key, payload):
if key in self.payloads:
return False
self.payloads[key] = dict(payload)
return True


def test_live_claim_is_create_only_and_permanent():
store = AtomicFakeStore()
kwargs = {
"store": store,
"account": "****1234",
"strategy_profile": "tqqq_core",
"run_period": "2026-08",
}

assert claim_live_strategy_run(**kwargs) is True
assert claim_live_strategy_run(**kwargs) is False
key = strategy_run_claim_key(
account="****1234", strategy_profile="tqqq_core", run_period="2026-08"
)
assert store.payloads[key]["stage"] == "PENDING_SUBMISSION"
assert store.payloads[key]["no_order_submitted"] is True
10 changes: 5 additions & 5 deletions uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.