LongBridgePlatform/strategy_registry.py at main · QuantStrategyLab/LongBridgePlatform · 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
124
125
126
127
128
129
130
131
from __future__ import annotations

from us_equity_strategies import (
    get_platform_runtime_adapter,
    get_runtime_enabled_profiles,
    get_strategy_catalog,
)

from quant_platform_kit.common.strategies import (
    PlatformCapabilityMatrix,
    PlatformStrategyPolicy,
    StrategyDefinition,
    StrategyMetadata,
    US_EQUITY_DOMAIN,
    build_platform_profile_matrix,
    build_platform_profile_status_matrix,
    derive_enabled_profiles_for_platform,
    derive_eligible_profiles_for_platform,
    get_catalog_strategy_metadata,
    get_enabled_profiles_for_platform,
    resolve_platform_strategy_definition,
)

LONGBRIDGE_PLATFORM = "longbridge"

LONGBRIDGE_ROLLOUT_ALLOWLIST = get_runtime_enabled_profiles()

PLATFORM_SUPPORTED_DOMAINS: dict[str, frozenset[str]] = {
    LONGBRIDGE_PLATFORM: frozenset({US_EQUITY_DOMAIN}),
}
STRATEGY_CATALOG = get_strategy_catalog()
PLATFORM_CAPABILITY_MATRIX = PlatformCapabilityMatrix(
    platform_id=LONGBRIDGE_PLATFORM,
    supported_domains=PLATFORM_SUPPORTED_DOMAINS[LONGBRIDGE_PLATFORM],
    supported_target_modes=frozenset({"weight", "value"}),
    supported_inputs=frozenset(
        {
            "benchmark_history",
            "market_history",
            "portfolio_snapshot",
            "derived_indicators",
            "feature_snapshot",
            "indicators",
            "account_state",
            "snapshot",
        }
    ),
    supported_capabilities=frozenset(),
)
ELIGIBLE_STRATEGY_PROFILES = derive_eligible_profiles_for_platform(
    STRATEGY_CATALOG,
    capability_matrix=PLATFORM_CAPABILITY_MATRIX,
    runtime_adapter_loader=lambda profile: get_platform_runtime_adapter(
        profile,
        platform_id=LONGBRIDGE_PLATFORM,
    ),
)
LONGBRIDGE_ENABLED_PROFILES = derive_enabled_profiles_for_platform(
    STRATEGY_CATALOG,
    capability_matrix=PLATFORM_CAPABILITY_MATRIX,
    runtime_adapter_loader=lambda profile: get_platform_runtime_adapter(
        profile,
        platform_id=LONGBRIDGE_PLATFORM,
    ),
    rollout_allowlist=LONGBRIDGE_ROLLOUT_ALLOWLIST,
)
PLATFORM_POLICY = PlatformStrategyPolicy(
    platform_id=LONGBRIDGE_PLATFORM,
    supported_domains=PLATFORM_SUPPORTED_DOMAINS[LONGBRIDGE_PLATFORM],
    enabled_profiles=LONGBRIDGE_ENABLED_PROFILES,
    default_profile="",
    rollback_profile="",
    require_explicit_profile=True,
)

SUPPORTED_STRATEGY_PROFILES = LONGBRIDGE_ENABLED_PROFILES
_SELECTION_ROLE_FIELDS = frozenset({"is_default", "is_rollback"})


def _without_selection_role_fields(row: dict[str, object]) -> dict[str, object]:
    return {key: value for key, value in row.items() if key not in _SELECTION_ROLE_FIELDS}


def get_eligible_profiles_for_platform(platform_id: str) -> frozenset[str]:
    if platform_id != LONGBRIDGE_PLATFORM:
        return frozenset()
    return ELIGIBLE_STRATEGY_PROFILES


def get_supported_profiles_for_platform(platform_id: str) -> frozenset[str]:
    return get_enabled_profiles_for_platform(platform_id, policy=PLATFORM_POLICY)


def get_platform_profile_matrix() -> list[dict[str, object]]:
    return [
        _without_selection_role_fields(row)
        for row in build_platform_profile_matrix(STRATEGY_CATALOG, policy=PLATFORM_POLICY)
    ]


def get_platform_profile_status_matrix() -> list[dict[str, object]]:
    return [
        _without_selection_role_fields(row)
        for row in build_platform_profile_status_matrix(
            STRATEGY_CATALOG,
            policy=PLATFORM_POLICY,
            eligible_profiles=ELIGIBLE_STRATEGY_PROFILES,
        )
    ]


def resolve_strategy_definition(
    raw_value: str | None,
    *,
    platform_id: str,
) -> StrategyDefinition:
    return resolve_platform_strategy_definition(
        raw_value,
        platform_id=platform_id,
        strategy_catalog=STRATEGY_CATALOG,
        policy=PLATFORM_POLICY,
    )


def resolve_strategy_metadata(
    raw_value: str | None,
    *,
    platform_id: str,
) -> StrategyMetadata:
    definition = resolve_strategy_definition(raw_value, platform_id=platform_id)
    return get_catalog_strategy_metadata(STRATEGY_CATALOG, definition.profile)