refactor: expose unified crypto strategy entrypoints

.github/workflows/ci.yml

@@ -13,6 +13,23 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v6
 
+      - name: Resolve QuantPlatformKit ref
+        id: quant-platform-kit-ref
+        run: |
+          set -euo pipefail
+          ref="main"
+          if [ -n "${GITHUB_HEAD_REF:-}" ] && git ls-remote --exit-code --heads https://github.com/QuantStrategyLab/QuantPlatformKit.git "${GITHUB_HEAD_REF}" >/dev/null 2>&1; then
+            ref="${GITHUB_HEAD_REF}"
+          fi
+          echo "ref=${ref}" >> "$GITHUB_OUTPUT"
+
+      - name: Checkout QuantPlatformKit
+        uses: actions/checkout@v6
+        with:
+          repository: QuantStrategyLab/QuantPlatformKit
+          ref: ${{ steps.quant-platform-kit-ref.outputs.ref }}
+          path: external/QuantPlatformKit
+
       - name: Setup Python
         uses: actions/setup-python@v6
         with:
@@ -22,7 +39,8 @@ jobs:
         run: |
           set -euo pipefail
           python -m pip install --upgrade pip
-          python -m pip install -e . ruff
+          python -m pip install -e . numpy pandas ruff
+          python -m pip install --no-deps -e external/QuantPlatformKit
 
       - name: Run Ruff
         run: |

pyproject.toml

@@ -9,7 +9,7 @@ description = "Shared crypto strategy catalog and implementations"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@6e8cc058b821aea8a54015d4b39e02fbdd3dc198",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@5174d9e40f79fffae47450a42e26434145d28b31",
 ]
 
 [tool.setuptools]

scripts/prepare_auto_optimization_pr.py

@@ -97,7 +97,8 @@
         r"^market_snapshot_support\.py$",
         r"^trade_state_support\.py$",
         r"^trend_pool_support\.py$",
-        r"^strategy_core\.py$",
+        r"^strategy_runtime\.py$",
+        r"^decision_mapper\.py$",
         r"^strategy_loader\.py$",
         r"^strategy_registry\.py$",
     ),

src/crypto_strategies/__init__.py

@@ -4,6 +4,7 @@
     get_strategy_catalog,
     get_strategy_definition,
     get_strategy_definitions,
+    get_strategy_entrypoint,
     get_strategy_index_rows,
     get_strategy_metadata,
 )
@@ -14,6 +15,7 @@
     "get_strategy_catalog",
     "get_strategy_definition",
     "get_strategy_definitions",
+    "get_strategy_entrypoint",
     "get_strategy_index_rows",
     "get_strategy_metadata",
 ]

src/crypto_strategies/catalog.py

@@ -5,15 +5,31 @@
     StrategyCatalog,
     StrategyComponentDefinition,
     StrategyDefinition,
+    StrategyEntrypointDefinition,
     StrategyMetadata,
     build_strategy_catalog,
     build_strategy_index_rows,
     get_catalog_strategy_definition,
     get_catalog_strategy_metadata,
+    load_strategy_entrypoint,
 )
 
 CRYPTO_LEADER_ROTATION_PROFILE = "crypto_leader_rotation"
 
+CRYPTO_LEADER_ROTATION_DEFAULT_CONFIG = {
+    "trend_pool_size": 5,
+    "rotation_top_n": 2,
+    "min_history_days": 365,
+    "min_avg_quote_vol_180": 8000000.0,
+    "membership_bonus": 0.10,
+    "weight_mode": "inverse_vol",
+    "allow_rotation_refresh": True,
+    "atr_multiplier": 2.5,
+    "artifact_contract_version": "crypto_leader_rotation.live_pool.v1",
+    "artifact_max_age_days": 45,
+    "artifact_acceptable_modes": ("core_major",),
+}
+
 STRATEGY_DEFINITIONS: dict[str, StrategyDefinition] = {
     CRYPTO_LEADER_ROTATION_PROFILE: StrategyDefinition(
         profile=CRYPTO_LEADER_ROTATION_PROFILE,
@@ -29,6 +45,20 @@
                 module_path="crypto_strategies.strategies.crypto_leader_rotation.rotation",
             ),
         ),
+        entrypoint=StrategyEntrypointDefinition(
+            module_path="crypto_strategies.entrypoints",
+            attribute_name="crypto_leader_rotation_entrypoint",
+        ),
+        required_inputs=frozenset(
+            {
+                "prices",
+                "trend_indicators",
+                "btc_snapshot",
+                "account_metrics",
+                "trend_universe_symbols",
+            }
+        ),
+        default_config=CRYPTO_LEADER_ROTATION_DEFAULT_CONFIG,
     ),
 }
 
@@ -64,6 +94,12 @@ def get_strategy_definition(profile: str) -> StrategyDefinition:
     return get_catalog_strategy_definition(STRATEGY_CATALOG, profile)
 
 
+def get_strategy_entrypoint(profile: str):
+    definition = get_strategy_definition(profile)
+    metadata = get_strategy_metadata(profile)
+    return load_strategy_entrypoint(definition, metadata=metadata)
+
+
 def get_strategy_metadata(profile: str) -> StrategyMetadata:
     return get_catalog_strategy_metadata(STRATEGY_CATALOG, profile)
 

src/crypto_strategies/entrypoints/__init__.py

@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from copy import deepcopy
+
+from quant_platform_kit.strategy_contracts import (
+    BudgetIntent,
+    CallableStrategyEntrypoint,
+    PositionTarget,
+    StrategyContext,
+    StrategyDecision,
+)
+
+from crypto_strategies.manifests import crypto_leader_rotation_manifest
+from crypto_strategies.strategies.crypto_leader_rotation import core as legacy_core
+from crypto_strategies.strategies.crypto_leader_rotation import rotation as legacy_rotation
+
+
+"""Unified crypto strategy entrypoints built on top of legacy core/rotation modules."""
+
+
+def _merge_runtime_config(ctx: StrategyContext) -> dict[str, object]:
+    config = dict(crypto_leader_rotation_manifest.default_config)
+    config.update(dict(ctx.runtime_config))
+    return config
+
+
+def _require_market_data(ctx: StrategyContext, key: str):
+    if key not in ctx.market_data:
+        raise ValueError(f"StrategyContext.market_data missing required key: {key}")
+    return ctx.market_data[key]
+
+
+def _resolve_translator(config: Mapping[str, object]) -> Callable[[str], str]:
+    translator = config.get("translator")
+    if callable(translator):
+        return translator
+    return lambda key, **_kwargs: str(key)
+
+
+def _default_symbol_state() -> dict[str, object]:
+    return {
+        "is_holding": False,
+        "entry_price": 0.0,
+        "highest_price": 0.0,
+    }
+
+
+def _resolve_state_helpers(config: Mapping[str, object]):
+    get_symbol_trade_state_fn = config.get("get_symbol_trade_state_fn")
+    set_symbol_trade_state_fn = config.get("set_symbol_trade_state_fn")
+
+    if callable(get_symbol_trade_state_fn) and callable(set_symbol_trade_state_fn):
+        return get_symbol_trade_state_fn, set_symbol_trade_state_fn
+
+    def _get_symbol_trade_state(state, symbol):
+        value = state.get(symbol)
+        if not isinstance(value, Mapping):
+            return _default_symbol_state()
+        merged = _default_symbol_state()
+        merged.update(dict(value))
+        return merged
+
+    def _set_symbol_trade_state(state, symbol, symbol_state):
+        state[symbol] = dict(symbol_state)
+
+    return _get_symbol_trade_state, _set_symbol_trade_state
+
+
+def evaluate_crypto_leader_rotation(ctx: StrategyContext) -> StrategyDecision:
+    config = _merge_runtime_config(ctx)
+    prices = _require_market_data(ctx, "prices")
+    indicators_map = _require_market_data(ctx, "trend_indicators")
+    btc_snapshot = _require_market_data(ctx, "btc_snapshot")
+    account_metrics = _require_market_data(ctx, "account_metrics")
+    trend_universe_symbols = list(_require_market_data(ctx, "trend_universe_symbols"))
+    state = dict(ctx.state)
+    working_state = deepcopy(state)
+    translator = _resolve_translator(config)
+    get_symbol_trade_state_fn, set_symbol_trade_state_fn = _resolve_state_helpers(config)
+
+    budgets = legacy_core.compute_allocation_budgets(
+        account_metrics["total_equity"],
+        account_metrics["cash_usdt"],
+        account_metrics.get("trend_value", 0.0),
+        account_metrics.get("dca_value", 0.0),
+    )
+
+    selected_pool, ranking = legacy_rotation.refresh_rotation_pool(
+        working_state,
+        indicators_map,
+        btc_snapshot,
+        trend_universe_symbols=trend_universe_symbols,
+        trend_pool_size=config["trend_pool_size"],
+        build_stable_quality_pool_fn=lambda indicators, btc, previous_pool: legacy_core.build_stable_quality_pool(
+            indicators,
+            btc,
+            previous_pool,
+            pool_size=config["trend_pool_size"],
+            min_history_days=config["min_history_days"],
+            min_avg_quote_vol_180=config["min_avg_quote_vol_180"],
+            membership_bonus=config["membership_bonus"],
+        ),
+        allow_refresh=bool(config.get("allow_rotation_refresh", True)),
+        now_utc=config.get("now_utc"),
+    )
+    selected_candidates = legacy_core.select_rotation_weights(
+        indicators_map,
+        prices,
+        btc_snapshot,
+        selected_pool,
+        config["rotation_top_n"],
+        weight_mode=str(config.get("weight_mode", "inverse_vol")),
+    )
+
+    sell_reasons: dict[str, str] = {}
+    atr_multiplier = float(config.get("atr_multiplier", 2.5))
+    for symbol in trend_universe_symbols:
+        curr_price = prices.get(symbol)
+        if curr_price is None:
+            continue
+        reason = legacy_rotation.get_trend_sell_reason(
+            working_state,
+            symbol,
+            curr_price,
+            indicators_map.get(symbol),
+            selected_candidates,
+            atr_multiplier,
+            get_symbol_trade_state_fn=get_symbol_trade_state_fn,
+            set_symbol_trade_state_fn=set_symbol_trade_state_fn,
+            translate_fn=translator,
+        )
+        if reason:
+            sell_reasons[symbol] = str(reason)
+
+    eligible_buy_symbols, planned_trend_buys = legacy_rotation.plan_trend_buys(
+        working_state,
+        runtime_trend_universe={symbol: {"base_asset": symbol[:-4]} for symbol in trend_universe_symbols},
+        selected_candidates=selected_candidates,
+        trend_indicators=indicators_map,
+        prices=prices,
+        available_trend_buy_budget=float(budgets["trend_usdt_pool"]),
+        allow_new_trend_entries=bool(config.get("allow_new_trend_entries", True)),
+        get_symbol_trade_state_fn=get_symbol_trade_state_fn,
+        allocate_trend_buy_budget_fn=legacy_core.allocate_trend_buy_budget,
+    )
+
+    positions = [
+        PositionTarget(
+            symbol="BTCUSDT",
+            target_weight=float(budgets["btc_target_ratio"]),
+            role="core",
+        )
+    ]
+    trend_target_ratio = float(budgets["trend_target_ratio"])
+    for symbol, payload in sorted(selected_candidates.items()):
+        positions.append(
+            PositionTarget(
+                symbol=symbol,
+                target_weight=trend_target_ratio * float(payload["weight"]),
+                role="trend_rotation",
+            )
+        )
+
+    budget_intents = (
+        BudgetIntent(
+            name="btc_core_dca_pool",
+            symbol="BTCUSDT",
+            amount=float(budgets["dca_usdt_pool"]),
+            purpose="btc_core_accumulation",
+        ),
+        BudgetIntent(
+            name="trend_rotation_pool",
+            amount=float(budgets["trend_usdt_pool"]),
+            purpose="trend_rotation",
+        ),
+    )
+
+    risk_flags: tuple[str, ...] = ()
+    if not btc_snapshot.get("regime_on"):
+        risk_flags += ("regime_off",)
+    if not selected_candidates:
+        risk_flags += ("no_trend_candidates",)
+
+    diagnostics = {
+        "trend_pool": tuple(selected_pool),
+        "rotation_candidates": {
+            symbol: {
+                "weight": float(payload["weight"]),
+                "relative_score": float(payload["relative_score"]),
+                "abs_momentum": float(payload["abs_momentum"]),
+            }
+            for symbol, payload in selected_candidates.items()
+        },
+        "ranking_preview": tuple(item["symbol"] for item in ranking[: int(config["trend_pool_size"])]),
+        "rotation_pool_source_version": working_state.get("rotation_pool_source_version"),
+        "rotation_pool_source_as_of_date": working_state.get("rotation_pool_source_as_of_date"),
+        "rotation_pool_last_month": working_state.get("rotation_pool_last_month"),
+        "sell_reasons": sell_reasons,
+        "eligible_buy_symbols": tuple(eligible_buy_symbols),
+        "planned_trend_buys": {symbol: float(amount) for symbol, amount in planned_trend_buys.items()},
+        "btc_base_order_usdt": float(legacy_core.get_dynamic_btc_base_order(account_metrics["total_equity"])),
+        "btc_target_ratio": float(budgets["btc_target_ratio"]),
+        "trend_target_ratio": float(budgets["trend_target_ratio"]),
+        "artifact_contract": {
+            "version": config.get("artifact_contract_version"),
+            "max_age_days": config.get("artifact_max_age_days"),
+            "acceptable_modes": tuple(config.get("artifact_acceptable_modes", ())),
+            **dict(ctx.artifacts.get("trend_pool_contract", {})),
+        },
+    }
+    return StrategyDecision(
+        positions=tuple(positions),
+        budgets=budget_intents,
+        risk_flags=risk_flags,
+        diagnostics=diagnostics,
+    )
+
+
+crypto_leader_rotation_entrypoint = CallableStrategyEntrypoint(
+    manifest=crypto_leader_rotation_manifest,
+    _evaluate=evaluate_crypto_leader_rotation,
+)
+
+
+__all__ = ["crypto_leader_rotation_entrypoint", "evaluate_crypto_leader_rotation"]