Skip to content

[codex] Align LongBridge strategy runtime contracts #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 1 commit into from
Apr 16, 2026
Merged

[codex] Align LongBridge strategy runtime contracts #21

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
46 changes: 10 additions & 36 deletions .github/workflows/sync-cloud-run-env.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,10 +80,7 @@ jobs:
import os
import subprocess
import sys
from pathlib import Path

from quant_platform_kit.common.strategies import derive_strategy_artifact_paths
from us_equity_strategies import get_strategy_catalog, resolve_canonical_profile
from us_equity_strategies import resolve_canonical_profile

profile = os.environ.get("STRATEGY_PROFILE", "").strip().lower()
if not profile:
Expand All @@ -102,16 +99,6 @@ jobs:
if not selected.get("eligible") or not selected.get("enabled"):
raise SystemExit(f"STRATEGY_PROFILE={profile!r} is not eligible/enabled: {selected}")

artifact_paths = derive_strategy_artifact_paths(
get_strategy_catalog(),
canonical_profile,
repo_root=Path.cwd(),
)
has_bundled_strategy_config = bool(
artifact_paths.bundled_config_path is not None
and Path(artifact_paths.bundled_config_path).exists()
)

output_path = os.environ["GITHUB_OUTPUT"]
with open(output_path, "a", encoding="utf-8") as output:
output.write(
Expand All @@ -124,7 +111,7 @@ jobs:
f"requires_strategy_config_path={str(bool(selected.get('requires_strategy_config_path'))).lower()}\n"
)
output.write(
f"has_bundled_strategy_config={str(has_bundled_strategy_config).lower()}\n"
f"config_source_policy={str(selected.get('config_source_policy') or 'none')}\n"
)
PY

Expand All @@ -134,7 +121,7 @@ jobs:
REQUIRES_SNAPSHOT_ARTIFACTS: ${{ steps.strategy_requirements.outputs.requires_snapshot_artifacts }}
REQUIRES_SNAPSHOT_MANIFEST_PATH: ${{ steps.strategy_requirements.outputs.requires_snapshot_manifest_path }}
REQUIRES_STRATEGY_CONFIG_PATH: ${{ steps.strategy_requirements.outputs.requires_strategy_config_path }}
HAS_BUNDLED_STRATEGY_CONFIG: ${{ steps.strategy_requirements.outputs.has_bundled_strategy_config }}
CONFIG_SOURCE_POLICY: ${{ steps.strategy_requirements.outputs.config_source_policy }}
run: |
set -euo pipefail

Expand Down Expand Up @@ -175,8 +162,8 @@ jobs:
fi

if [ "${REQUIRES_STRATEGY_CONFIG_PATH:-}" = "true" ] \
&& [ -z "${LONGBRIDGE_STRATEGY_CONFIG_PATH:-}" ] \
&& [ "${HAS_BUNDLED_STRATEGY_CONFIG:-}" != "true" ]; then
&& [ "${CONFIG_SOURCE_POLICY:-}" = "env_only" ] \
&& [ -z "${LONGBRIDGE_STRATEGY_CONFIG_PATH:-}" ]; then
missing_vars+=("LONGBRIDGE_STRATEGY_CONFIG_PATH")
fi

Expand Down Expand Up @@ -390,10 +377,7 @@ jobs:
import os
import subprocess
import sys
from pathlib import Path

from quant_platform_kit.common.strategies import derive_strategy_artifact_paths
from us_equity_strategies import get_strategy_catalog, resolve_canonical_profile
from us_equity_strategies import resolve_canonical_profile

profile = os.environ.get("STRATEGY_PROFILE", "").strip().lower()
if not profile:
Expand All @@ -412,16 +396,6 @@ jobs:
if not selected.get("eligible") or not selected.get("enabled"):
raise SystemExit(f"STRATEGY_PROFILE={profile!r} is not eligible/enabled: {selected}")

artifact_paths = derive_strategy_artifact_paths(
get_strategy_catalog(),
canonical_profile,
repo_root=Path.cwd(),
)
has_bundled_strategy_config = bool(
artifact_paths.bundled_config_path is not None
and Path(artifact_paths.bundled_config_path).exists()
)

output_path = os.environ["GITHUB_OUTPUT"]
with open(output_path, "a", encoding="utf-8") as output:
output.write(
Expand All @@ -434,7 +408,7 @@ jobs:
f"requires_strategy_config_path={str(bool(selected.get('requires_strategy_config_path'))).lower()}\n"
)
output.write(
f"has_bundled_strategy_config={str(has_bundled_strategy_config).lower()}\n"
f"config_source_policy={str(selected.get('config_source_policy') or 'none')}\n"
)
PY

Expand All @@ -444,7 +418,7 @@ jobs:
REQUIRES_SNAPSHOT_ARTIFACTS: ${{ steps.strategy_requirements.outputs.requires_snapshot_artifacts }}
REQUIRES_SNAPSHOT_MANIFEST_PATH: ${{ steps.strategy_requirements.outputs.requires_snapshot_manifest_path }}
REQUIRES_STRATEGY_CONFIG_PATH: ${{ steps.strategy_requirements.outputs.requires_strategy_config_path }}
HAS_BUNDLED_STRATEGY_CONFIG: ${{ steps.strategy_requirements.outputs.has_bundled_strategy_config }}
CONFIG_SOURCE_POLICY: ${{ steps.strategy_requirements.outputs.config_source_policy }}
run: |
set -euo pipefail

Expand Down Expand Up @@ -485,8 +459,8 @@ jobs:
fi

if [ "${REQUIRES_STRATEGY_CONFIG_PATH:-}" = "true" ] \
&& [ -z "${LONGBRIDGE_STRATEGY_CONFIG_PATH:-}" ] \
&& [ "${HAS_BUNDLED_STRATEGY_CONFIG:-}" != "true" ]; then
&& [ "${CONFIG_SOURCE_POLICY:-}" = "env_only" ] \
&& [ -z "${LONGBRIDGE_STRATEGY_CONFIG_PATH:-}" ]; then
missing_vars+=("LONGBRIDGE_STRATEGY_CONFIG_PATH")
fi

Expand Down
87 changes: 18 additions & 69 deletions runtime_config_support.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,10 @@
from pathlib import Path
from typing import Callable

from quant_platform_kit.common.strategies import derive_strategy_artifact_paths
from quant_platform_kit.common.runtime_config import (
resolve_bool_value,
resolve_strategy_runtime_path_settings,
)
from strategy_registry import (
LONGBRIDGE_PLATFORM,
resolve_strategy_definition,
Expand Down Expand Up @@ -73,78 +76,38 @@ def load_platform_runtime_settings(
strategy_definition.profile,
platform_id=LONGBRIDGE_PLATFORM,
)
artifact_root = _first_non_empty(
os.getenv("LONGBRIDGE_STRATEGY_ARTIFACT_ROOT"),
os.getenv("STRATEGY_ARTIFACT_ROOT"),
)
derived_artifact_paths = derive_strategy_artifact_paths(
get_strategy_catalog(),
strategy_definition.profile,
artifact_root=artifact_root,
runtime_paths = resolve_strategy_runtime_path_settings(
strategy_catalog=get_strategy_catalog(),
strategy_definition=strategy_definition,
strategy_metadata=strategy_metadata,
platform_env_prefix="LONGBRIDGE",
env=os.environ,
repo_root=Path(__file__).resolve().parent,
)
strategy_config_path, strategy_config_source = resolve_strategy_config_path(
explicit_path=_first_non_empty(
os.getenv("LONGBRIDGE_STRATEGY_CONFIG_PATH"),
os.getenv("STRATEGY_CONFIG_PATH"),
),
bundled_path=(
str(derived_artifact_paths.bundled_config_path)
if derived_artifact_paths.bundled_config_path is not None
else None
),
)
return PlatformRuntimeSettings(
project_id=project_id_resolver(),
secret_name=os.getenv("LONGPORT_SECRET_NAME", DEFAULT_LONGPORT_SECRET_NAME),
account_prefix=account_prefix,
strategy_profile=strategy_definition.profile,
strategy_display_name=strategy_metadata.display_name,
strategy_domain=strategy_definition.domain,
strategy_profile=runtime_paths.strategy_profile,
strategy_display_name=runtime_paths.strategy_display_name,
strategy_domain=runtime_paths.strategy_domain,
account_region=infer_account_region(
os.getenv("ACCOUNT_REGION"),
account_prefix=account_prefix,
),
notify_lang=os.getenv("NOTIFY_LANG", "en"),
tg_token=os.getenv("TELEGRAM_TOKEN"),
tg_chat_id=os.getenv("GLOBAL_TELEGRAM_CHAT_ID"),
dry_run_only=_resolve_bool_env("LONGBRIDGE_DRY_RUN_ONLY"),
dry_run_only=resolve_bool_value(os.getenv("LONGBRIDGE_DRY_RUN_ONLY")),
income_threshold_usd=_optional_float_env("INCOME_THRESHOLD_USD"),
qqqi_income_ratio=_qqqi_income_ratio_env(),
feature_snapshot_path=_first_non_empty(
os.getenv("LONGBRIDGE_FEATURE_SNAPSHOT_PATH"),
os.getenv("FEATURE_SNAPSHOT_PATH"),
str(derived_artifact_paths.feature_snapshot_path)
if derived_artifact_paths.feature_snapshot_path is not None
else None,
),
feature_snapshot_manifest_path=_first_non_empty(
os.getenv("LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"),
os.getenv("FEATURE_SNAPSHOT_MANIFEST_PATH"),
str(derived_artifact_paths.feature_snapshot_manifest_path)
if derived_artifact_paths.feature_snapshot_manifest_path is not None
else None,
),
strategy_config_path=strategy_config_path,
strategy_config_source=strategy_config_source,
feature_snapshot_path=runtime_paths.feature_snapshot_path,
feature_snapshot_manifest_path=runtime_paths.feature_snapshot_manifest_path,
strategy_config_path=runtime_paths.strategy_config_path,
strategy_config_source=runtime_paths.strategy_config_source,
)


def resolve_strategy_config_path(
*,
explicit_path: str | None,
bundled_path: str | None,
) -> tuple[str | None, str | None]:
path = _first_non_empty(explicit_path)
if path is not None:
return path, "env"

bundled = _first_non_empty(bundled_path)
if bundled is not None and Path(bundled).exists():
return bundled, "bundled_canonical_default"
return None, None


def _normalize_region(raw_value: str | None) -> str | None:
if raw_value is None:
return None
Expand All @@ -154,13 +117,6 @@ def _normalize_region(raw_value: str | None) -> str | None:
return value.upper()


def _resolve_bool_env(name: str) -> bool:
raw_value = os.getenv(name)
if raw_value is None:
return False
return str(raw_value).strip().lower() in {"1", "true", "yes", "on"}


def _optional_float_env(name: str) -> float | None:
raw_value = os.getenv(name)
if raw_value is None or raw_value.strip() == "":
Expand All @@ -174,10 +130,3 @@ def _qqqi_income_ratio_env() -> float | None:
raise ValueError(f"QQQI_INCOME_RATIO must be in [0,1], got {value}")
return value


def _first_non_empty(*values: str | None) -> str | None:
for value in values:
text = str(value or "").strip()
if text:
return text
return None
20 changes: 17 additions & 3 deletions scripts/print_strategy_switch_env_plan.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,11 @@ def build_switch_plan(profile: str, *, account_region: str | None = None) -> dic
platform_id=LONGBRIDGE_PLATFORM,
)
requires_feature_snapshot = bool(runtime_requirements["requires_snapshot_artifacts"])
requires_snapshot_manifest_path = bool(
runtime_requirements["requires_snapshot_manifest_path"]
)
requires_strategy_config_path = bool(runtime_requirements["requires_strategy_config_path"])
config_source_policy = str(runtime_requirements.get("config_source_policy") or "none")
normalized_region = (account_region or "").strip().upper()

set_env: dict[str, str] = {"STRATEGY_PROFILE": definition.profile}
Expand All @@ -66,9 +70,19 @@ def build_switch_plan(profile: str, *, account_region: str | None = None) -> dic

if requires_feature_snapshot:
set_env["LONGBRIDGE_FEATURE_SNAPSHOT_PATH"] = "<required>"
set_env["LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"] = "<required>"
if requires_strategy_config_path and artifact_paths.bundled_config_path is not None:
set_env["LONGBRIDGE_STRATEGY_CONFIG_PATH"] = str(artifact_paths.bundled_config_path)
if requires_snapshot_manifest_path:
set_env["LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH"] = "<required>"
else:
remove_if_present.append("LONGBRIDGE_FEATURE_SNAPSHOT_MANIFEST_PATH")
if requires_strategy_config_path and config_source_policy == "env_only":
set_env["LONGBRIDGE_STRATEGY_CONFIG_PATH"] = "<required>"
elif requires_strategy_config_path and config_source_policy == "bundled_or_env":
remove_if_present.append("LONGBRIDGE_STRATEGY_CONFIG_PATH")
notes.append(
"LONGBRIDGE_STRATEGY_CONFIG_PATH is optional for bundled_or_env profiles; leave it unset to use the packaged canonical config."
)
elif requires_strategy_config_path:
set_env["LONGBRIDGE_STRATEGY_CONFIG_PATH"] = "<required>"
else:
remove_if_present.append("LONGBRIDGE_STRATEGY_CONFIG_PATH")
else:
Expand Down
Loading