Skip to content
Merged
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
21 changes: 15 additions & 6 deletions docs/b06_execution_presentation_facade.zh-CN.md
Original file line number Diff line number Diff line change
@@ -1,17 +1,26 @@
# B06:value-target 执行 API 与展示布局分离(第一刀)
# B06 / N05:value-target 执行 API 与展示布局分离

## 做了什么
## 已做

### 第一刀(QPK #654)
- 新增 `quant_platform_kit.common.presentation`:
- `ValueTargetExecutionSemantics`(阈值/时序/数值)
- `ValueTargetDisplayAnnotations`(文案)
- `ValueTargetPlanPresentation`(`portfolio_rows_layout` / `execution_fields`)
- 新增 `build_value_target_execution_runtime_plan`:**不**在签名上接收 dashboard 布局 kwargs
- 保留 `build_value_target_runtime_plan` 为兼容 facade(旧 kwargs → Presentation)

## 未做(follow-up)
- 未删除 `ValueTargetExecutionAnnotations` 混合字段;未迁移平台 `decision_mapper` 调用方
- 未把 SOXL 公式移出 `runtime_inputs.py`(属 B03/B07)
- 未改阈值/数量/时序数值行为
### 第二刀(本 PR:annotations 展示字段归属)
- 明确 `VALUE_TARGET_DISPLAY_FIELD_NAMES` 为展示字段真源
- 新增 `split_value_target_annotation_parts` / `resolve_value_target_execution_annotations`
- `build_value_target_execution_runtime_plan` 支持首选 `semantics=` + `display=`
- `build_value_target_execution_annotations` 内部改为 split→merge(行为等价)
- `ValueTargetExecutionAnnotations` **仍保留混合字段**作为 wire/compat aggregate(batch-upgrade-then-remove;删字段等 B12 采用证据)

## 未做
- 未从 `ValueTargetExecutionAnnotations` **删除**展示字段(B03:消费者迁移未齐前禁止)
- 未迁移 LongBridge / Firstrade `decision_mapper`(禁区)
- 未把 SOXL 公式移出 `runtime_inputs.py`(B03/B07)

## 验证
`pytest tests/test_strategy_contracts.py tests/test_value_target_presentation_facade.py`
21 changes: 17 additions & 4 deletions src/quant_platform_kit/common/execution_translation.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,12 @@
from dataclasses import dataclass
from typing import Any

from .presentation.value_target_plan import ValueTargetPlanPresentation
from .presentation.value_target_plan import (
ValueTargetDisplayAnnotations,
ValueTargetExecutionSemantics,
ValueTargetPlanPresentation,
resolve_value_target_execution_annotations,
)
from .strategy_contracts import (
PositionTarget,
StrategyContractValidationError,
Expand Down Expand Up @@ -228,21 +233,29 @@ def build_value_target_execution_runtime_plan(
portfolio_inputs: ValueTargetPortfolioInputs,
strategy_symbols_order: str = "risk_safe_income",
annotations: ValueTargetExecutionAnnotations | None = None,
semantics: ValueTargetExecutionSemantics | None = None,
display: ValueTargetDisplayAnnotations | None = None,
include_sellable_quantities: bool | None = None,
presentation: ValueTargetPlanPresentation | None = None,
) -> dict[str, Any]:
"""Build the value-target runtime plan without dashboard-layout kwargs.

Prefer ``semantics`` + optional ``display`` for new callers. ``annotations``
remains the compatibility aggregate (mixed execution + display fields).
Presentation knobs (portfolio row layout / field filters) are optional via
``ValueTargetPlanPresentation``. Callers that only need execution semantics
should omit ``presentation`` and use the default structural layout.
``ValueTargetPlanPresentation``.
"""
presentation = presentation or ValueTargetPlanPresentation()
execution_plan = build_value_target_execution_plan(
decision,
strategy_profile=strategy_profile,
)
resolved_annotations = annotations or build_value_target_execution_annotations(decision)
resolved_annotations = resolve_value_target_execution_annotations(
decision=decision,
annotations=annotations,
semantics=semantics,
display=display,
)
portfolio_plan = build_value_target_portfolio_plan(
execution_plan,
market_values=portfolio_inputs.market_values,
Expand Down
6 changes: 6 additions & 0 deletions src/quant_platform_kit/common/presentation/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,19 +6,25 @@
"""

from .value_target_plan import (
VALUE_TARGET_DISPLAY_FIELD_NAMES,
ValueTargetDisplayAnnotations,
ValueTargetExecutionSemantics,
ValueTargetPlanPresentation,
display_annotations_from_execution_annotations,
execution_semantics_from_annotations,
merge_value_target_execution_annotations,
resolve_value_target_execution_annotations,
split_value_target_annotation_parts,
)

__all__ = [
"VALUE_TARGET_DISPLAY_FIELD_NAMES",
"ValueTargetDisplayAnnotations",
"ValueTargetExecutionSemantics",
"ValueTargetPlanPresentation",
"display_annotations_from_execution_annotations",
"execution_semantics_from_annotations",
"merge_value_target_execution_annotations",
"resolve_value_target_execution_annotations",
"split_value_target_annotation_parts",
]
189 changes: 187 additions & 2 deletions src/quant_platform_kit/common/presentation/value_target_plan.py
Original file line number Diff line number Diff line change
@@ -1,19 +1,44 @@
"""Split value-target execution semantics from display/layout knobs.

``ValueTargetExecutionAnnotations`` in ``strategy_contracts`` remains the
compatibility wire type (all fields). New callers should prefer:
compatibility wire aggregate (execution + display fields). New callers should
prefer:
- ``ValueTargetExecutionSemantics`` for thresholds / timing / numeric gates
- ``ValueTargetDisplayAnnotations`` for dashboard/signal copy
- ``ValueTargetPlanPresentation`` for portfolio row layout / field selection

Construct the wire aggregate with ``merge_value_target_execution_annotations``
(or pass ``semantics`` / ``display`` into
``build_value_target_execution_runtime_plan``). Mixed-field removal waits on
consumer adoption evidence (B12).
"""

from __future__ import annotations

import math
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any

from quant_platform_kit.common.strategy_contracts import ValueTargetExecutionAnnotations
from quant_platform_kit.common.strategy_contracts import (
StrategyContractValidationError,
StrategyDecision,
ValueTargetExecutionAnnotations,
validate_strategy_decision,
)

# Source of truth for which annotation keys are presentation-owned.
VALUE_TARGET_DISPLAY_FIELD_NAMES: frozenset[str] = frozenset(
{
"signal_display",
"status_display",
"dashboard_text",
"separator",
"deploy_ratio_text",
"income_ratio_text",
"income_locked_ratio_text",
}
)


@dataclass(frozen=True)
Expand Down Expand Up @@ -97,6 +122,7 @@ def merge_value_target_execution_annotations(
semantics: ValueTargetExecutionSemantics,
display: ValueTargetDisplayAnnotations | None = None,
) -> ValueTargetExecutionAnnotations:
"""Build the compatibility wire aggregate from split types."""
display = display or ValueTargetDisplayAnnotations()
return ValueTargetExecutionAnnotations(
trade_threshold_value=float(semantics.trade_threshold_value),
Expand All @@ -121,3 +147,162 @@ def merge_value_target_execution_annotations(
current_min_trade=semantics.current_min_trade,
investable_cash=semantics.investable_cash,
)


def _pick_annotation_str(
annotations: Mapping[str, Any],
diagnostics: Mapping[str, Any],
*keys: str,
) -> str | None:
for key in keys:
value = annotations.get(key, diagnostics.get(key))
if value is None:
continue
text = str(value).strip()
if text:
return text
return None


def _ensure_finite_annotation_number(value: object, *, field_name: str) -> None:
if value is None:
return
if not isinstance(value, (int, float)) or not math.isfinite(float(value)):
raise StrategyContractValidationError(
f"{field_name} must be a finite number when provided"
)


def _pick_annotation_float(
annotations: Mapping[str, Any],
diagnostics: Mapping[str, Any],
*keys: str,
default: float | None = None,
field_prefix: str = "execution_annotations",
) -> float | None:
for key in keys:
value = annotations.get(key, diagnostics.get(key))
if value is None:
continue
_ensure_finite_annotation_number(value, field_name=f"{field_prefix}.{key}")
return float(value)
return default


def split_value_target_annotation_parts(
decision: StrategyDecision,
) -> tuple[ValueTargetExecutionSemantics, ValueTargetDisplayAnnotations]:
"""Parse decision diagnostics into execution semantics + display copy."""
validate_strategy_decision(decision)
diagnostics = dict(decision.diagnostics)
raw_annotations = diagnostics.get("execution_annotations")
annotations = dict(raw_annotations) if isinstance(raw_annotations, Mapping) else {}

threshold_value = _pick_annotation_float(
annotations,
diagnostics,
"trade_threshold_value",
"threshold",
"threshold_value",
)
if threshold_value is None:
raise StrategyContractValidationError(
"ValueTargetExecutionAnnotations requires trade_threshold_value "
"(or legacy threshold/threshold_value)"
)

signal_delay = _pick_annotation_float(
annotations,
diagnostics,
"signal_effective_after_trading_days",
)
semantics = ValueTargetExecutionSemantics(
trade_threshold_value=threshold_value,
reserved_cash=float(
_pick_annotation_float(
annotations,
diagnostics,
"reserved_cash",
"reserved",
default=0.0,
)
or 0.0
),
signal_date=_pick_annotation_str(annotations, diagnostics, "signal_date"),
effective_date=_pick_annotation_str(annotations, diagnostics, "effective_date"),
execution_timing_contract=_pick_annotation_str(
annotations, diagnostics, "execution_timing_contract"
),
execution_calendar_source=_pick_annotation_str(
annotations, diagnostics, "execution_calendar_source"
),
signal_effective_after_trading_days=(
int(signal_delay) if signal_delay is not None else None
),
benchmark_symbol=_pick_annotation_str(annotations, diagnostics, "benchmark_symbol"),
benchmark_price=_pick_annotation_float(
annotations, diagnostics, "benchmark_price", "qqq_price"
),
long_trend_value=_pick_annotation_float(
annotations, diagnostics, "long_trend_value", "ma200"
),
exit_line=_pick_annotation_float(annotations, diagnostics, "exit_line"),
active_risk_asset=_pick_annotation_str(annotations, diagnostics, "active_risk_asset"),
current_min_trade=_pick_annotation_float(
annotations, diagnostics, "current_min_trade"
),
investable_cash=_pick_annotation_float(annotations, diagnostics, "investable_cash"),
)
display = ValueTargetDisplayAnnotations(
signal_display=_pick_annotation_str(
annotations, diagnostics, "signal_display", "signal_message"
),
status_display=_pick_annotation_str(
annotations, diagnostics, "status_display", "market_status"
),
dashboard_text=_pick_annotation_str(
annotations, diagnostics, "dashboard_text", "dashboard"
),
separator=_pick_annotation_str(annotations, diagnostics, "separator"),
deploy_ratio_text=_pick_annotation_str(
annotations, diagnostics, "deploy_ratio_text"
),
income_ratio_text=_pick_annotation_str(
annotations, diagnostics, "income_ratio_text"
),
income_locked_ratio_text=_pick_annotation_str(
annotations, diagnostics, "income_locked_ratio_text"
),
)
return semantics, display


def resolve_value_target_execution_annotations(
*,
decision: StrategyDecision | None = None,
annotations: ValueTargetExecutionAnnotations | None = None,
semantics: ValueTargetExecutionSemantics | None = None,
display: ValueTargetDisplayAnnotations | None = None,
) -> ValueTargetExecutionAnnotations:
"""Resolve wire annotations from preferred split types or legacy aggregate.

Precedence:
1. Explicit ``annotations`` (compat facade)
2. ``semantics`` (+ optional ``display``) merged
3. Parse ``decision`` diagnostics into split parts then merge
"""
if annotations is not None and (semantics is not None or display is not None):
raise StrategyContractValidationError(
"Pass either annotations= (compat) or semantics=/display= (preferred), not both"
)
if annotations is not None:
return annotations
if semantics is not None:
return merge_value_target_execution_annotations(semantics, display)
if decision is None:
raise StrategyContractValidationError(
"resolve_value_target_execution_annotations requires annotations, "
"semantics, or decision"
)
parsed_semantics, parsed_display = split_value_target_annotation_parts(decision)
return merge_value_target_execution_annotations(parsed_semantics, parsed_display)
Loading
Loading