QmtPlatform/strategy_runtime.py at main · QuantStrategyLab/QmtPlatform · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Mapping

from quant_platform_kit.common.feature_snapshot import load_feature_snapshot_guarded
from quant_platform_kit.common.feature_snapshot_runtime import (
    FeatureSnapshotRuntimeSettings,
    evaluate_feature_snapshot_strategy,
)
from quant_platform_kit.strategy_contracts import (
    StrategyDecision,
    StrategyEntrypoint,
    StrategyRuntimeAdapter,
    apply_runtime_policy_to_runtime_config,
    build_execution_timing_metadata,
    build_strategy_context_from_available_inputs,
)
from runtime_config_support import PlatformRuntimeSettings
from strategy_loader import (
    load_strategy_entrypoint_for_profile,
    load_strategy_runtime_adapter_for_profile,
)

_FEATURE_SNAPSHOT_INPUT = "feature_snapshot"


@dataclass(frozen=True)
class StrategyEvaluationResult:
    decision: StrategyDecision
    metadata: Mapping[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class LoadedStrategyRuntime:
    entrypoint: StrategyEntrypoint
    runtime_adapter: StrategyRuntimeAdapter
    runtime_settings: PlatformRuntimeSettings
    runtime_config: Mapping[str, Any] = field(default_factory=dict)

    @property
    def profile(self) -> str:
        return self.entrypoint.manifest.profile

    def evaluate(self, *, available_inputs: Mapping[str, Any]) -> StrategyEvaluationResult:
        runtime_config = dict(self.runtime_config)
        apply_runtime_policy_to_runtime_config(runtime_config, self.runtime_adapter)

        stamped_inputs = dict(available_inputs)
        snapshot = stamped_inputs.get("portfolio_snapshot")
        if snapshot is not None:
            from quant_platform_kit.strategy_lifecycle.live_equity import stamp_consecutive_losses_on_snapshot

            stamped_inputs["portfolio_snapshot"] = stamp_consecutive_losses_on_snapshot(
                snapshot,
                strategy_profile=self.profile,
            )

        if _FEATURE_SNAPSHOT_INPUT in frozenset(self.entrypoint.manifest.required_inputs):
            return self._evaluate_feature_snapshot_strategy(
                runtime_config=runtime_config,
                available_inputs=stamped_inputs,
            )

        as_of = datetime.now(timezone.utc)
        ctx = build_strategy_context_from_available_inputs(
            entrypoint=self.entrypoint,
            runtime_adapter=self.runtime_adapter,
            as_of=as_of,
            available_inputs=stamped_inputs,
            runtime_config=runtime_config,
        )
        decision = self.entrypoint.evaluate(ctx)
        return StrategyEvaluationResult(
            decision=decision,
            metadata={
                "strategy_profile": self.profile,
                **build_execution_timing_metadata(
                    signal_date=as_of,
                    signal_effective_after_trading_days=(
                        self.runtime_adapter.runtime_policy.signal_effective_after_trading_days
                    ),
                ),
            },
        )

    def _evaluate_feature_snapshot_strategy(
        self,
        *,
        runtime_config: Mapping[str, Any],
        available_inputs: Mapping[str, Any],
    ) -> StrategyEvaluationResult:
        runtime_config = dict(runtime_config)
        runtime_config.setdefault("run_as_of", datetime.now(timezone.utc).replace(tzinfo=None))
        result = evaluate_feature_snapshot_strategy(
            entrypoint=self.entrypoint,
            runtime_adapter=self.runtime_adapter,
            runtime_settings=FeatureSnapshotRuntimeSettings(
                feature_snapshot_path=self.runtime_settings.feature_snapshot_path,
                feature_snapshot_manifest_path=self.runtime_settings.feature_snapshot_manifest_path,
                dry_run_only=self.runtime_settings.dry_run_only,
            ),
            runtime_config=runtime_config,
            merged_runtime_config=dict(self.entrypoint.manifest.default_config),
            available_inputs=available_inputs,
            base_managed_symbols=(),
            include_strategy_display_name=True,
            set_run_as_of=True,
            snapshot_loader=load_feature_snapshot_guarded,
        )
        return StrategyEvaluationResult(decision=result.decision, metadata=result.metadata)


def load_strategy_runtime(raw_profile: str | None, *, runtime_settings: PlatformRuntimeSettings) -> LoadedStrategyRuntime:
    entrypoint = load_strategy_entrypoint_for_profile(raw_profile)
    runtime_adapter = load_strategy_runtime_adapter_for_profile(raw_profile)
    return LoadedStrategyRuntime(
        entrypoint=entrypoint,
        runtime_adapter=runtime_adapter,
        runtime_settings=runtime_settings,
        runtime_config=dict(entrypoint.manifest.default_config),
    )