Merge pull request #75 from fourblocks2020/copilot/feat-alerts-notify-reconcile-warned
feat(alerts): emit first `reconcile_warned` intent rejection alarm from `IntentService`
This commit is contained in:
@@ -0,0 +1,22 @@
|
||||
---
|
||||
id: "ac-0025"
|
||||
status: active
|
||||
---
|
||||
|
||||
# 재조정 경고에 따른 전략 주문 요청 최초 거부 알림
|
||||
|
||||
## Criterion
|
||||
|
||||
`trex-core`는 해당 slot에 해제되지 않은 재조정 경고가 남아 있어 `ReconcileWarningStore.is_warned(slot_id)`가 참을 반환하고, 그 때문에 외부 전략의 새로운 매수 또는 매도 요청을 `status="rejected"`, `reason="reconcile_warned"`로 거부한 경우, 그 결과를 `IntentStore.finalize()`에 저장한 뒤 해당 `intent_id`에 대한 경고를 한 건 발행해야 한다. 경고는 `trex.alerts`로 보내며 `severity="warn"`, `code="strategy_intent_rejected"`를 사용한다. 같은 `intent_id`가 다시 들어오거나 최초 요청을 처리하는 도중 중복 요청이 들어오더라도 경고를 다시 발행해서는 안 된다. 반면, 같은 slot을 대상으로 하더라도 `intent_id`가 다르면 각각 별개의 최초 거부로 보고 경고를 발행해야 한다.
|
||||
|
||||
경고에는 `instance_key`, `intent_id`, `intent_type`, `reason="reconcile_warned"`, `broker_submitted=false`를 반드시 포함한다. `slot_id`, `portfolio_id`, `symbol`, `strategy_id`는 신뢰할 수 있는 값이 확인된 경우에만 포함하며, 알 수 없는 값을 추정하거나 원본 요청에서 그대로 복사해서는 안 된다. 매수 요청의 `strategy_id`와 매도 요청의 `sell_strategy_id`는 경고에서 공통 필드인 `strategy_id`로 표현한다.
|
||||
|
||||
정상 승인된 요청, `reconcile_warned` 이외의 사유로 거부된 요청, 최종 결과 저장에 실패한 요청, 그리고 safety-sell·kill-switch·watchdog 같은 내부 안전 경로에서는 이 경고를 발행하지 않는다. 경고 발행기가 없거나 부가 정보를 조회하지 못하거나 메시지 전송이 실패하더라도 이미 결정된 거부 결과, RPC 응답, 재조정 경고, slot 상태 및 broker·ledger·portfolio-hub 동작을 중단하거나 변경해서는 안 된다. 경고와 이 처리 경로의 로그에는 계좌 식별정보, credential, token, password, DSN, 원본 요청, 전략의 `extra`, 외부 시스템의 원본 payload 또는 예외 원문을 포함해서는 안 된다.
|
||||
|
||||
## Verification
|
||||
|
||||
- 정상 동작: 재조정 경고가 설정된 slot에 새로운 매수 요청과 매도 요청을 각각 제출하면, 최종 거부 결과가 저장된 뒤 `strategy_intent_rejected` 경고가 `warn` 등급으로 요청마다 한 건씩 발행된다. 각 경고에는 필수 필드와 `broker_submitted=false`가 들어 있다. 매수의 `strategy_id`와 매도의 `sell_strategy_id`는 값이 있을 때 경고의 `strategy_id`로 나타나며, `slot_id`, `portfolio_id`, `symbol`은 저장소에서 확인된 값만 포함된다.
|
||||
- 중복 및 제외 조건: 같은 `intent_id`를 같은 프로세스에서 반복하거나 기존 영속 intent 저장소로 서비스를 다시 구성한 뒤 반복해도 경고는 추가로 발행되지 않고 저장된 기존 결과가 그대로 반환된다. 재조정 경고가 설정된 같은 slot에 서로 다른 `intent_id` 두 건을 제출하면 경고는 두 건 발행된다. 최초 요청을 처리하는 도중 들어온 중복 요청과 `duplicate_in_flight`, 정상 승인, `reconcile_warning_unavailable`, `capacity_full`, 입력 검증 실패 등 `reconcile_warned`가 아닌 결과는 경고를 만들지 않는다.
|
||||
- 처리 순서 및 장애 조건: 호출 순서를 기록하는 테스트용 저장소와 경고 발행기를 사용해 `IntentStore.finalize()`가 경고 발행보다 먼저 완료되는지 확인한다. 최종 결과 저장이 실패하면 경고를 발행하지 않는다. 경고 발행기가 없거나 실패를 반환하거나 전송 계층에 장애가 발생해도 원래의 `reconcile_warned` 거부 결과와 RPC 응답 본문은 유지된다. 이 과정에서 broker 주문 제출·취소, slot 상태 전이, 재조정 경고 설정·해제, ledger 기록 또는 portfolio-hub 쓰기가 추가로 발생하지 않는다.
|
||||
- 보안 및 호환성: 원본 요청과 `extra`에 credential·token·account·DSN·exception 식별용 값을 넣더라도 경고 제목, 필드, 직렬화된 전송 내용과 출력 로그에 해당 값이나 금지된 key가 나타나지 않는다. 부가 정보를 읽지 못하거나 값이 없으면 해당 선택 필드만 생략하고 필수 필드와 거부 결과는 유지한다. 운영 daemon의 `IntentService`에는 기존 공용 `AlarmEmitter`가 연결되며, 경고 발행기 없이 `IntentService`를 직접 생성하는 기존 사용 방식도 계속 동작한다.
|
||||
- 검사 방법: `tests/application/test_reconcile_warning_blocking.py` 또는 별도의 intent 경고 테스트에 매수·매도, 같은 `intent_id`와 다른 `intent_id`, 재시작을 모사할 수 있는 intent 저장소, 처리 순서, 발행 실패, 경고 필드 허용 목록과 부작용 없음 조건을 추가한다. 이어서 `tests/application/test_messaging_intent_consumer.py`, `tests/application/test_alarm_emitter.py`, daemon 연결 테스트, AC-0023·AC-0024 관련 회귀 테스트, 전체 저장소 테스트, Ruff, ProofLine validation과 `git diff --check`를 실행한다.
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"schema_version": 1,
|
||||
"next_line_number": 8,
|
||||
"next_ac_number": 25
|
||||
"next_line_number": 9,
|
||||
"next_ac_number": 26
|
||||
}
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
---
|
||||
id: "dcy-0008"
|
||||
status: confirmed
|
||||
---
|
||||
|
||||
# Reconcile-warned strategy intent rejection alerts
|
||||
|
||||
## Problem
|
||||
|
||||
활성 reconciliation warning이 있는 slot에 외부 strategy가 신규 buy 또는 sell intent를 제출하면 `IntentService`는 `reconcile_warned` 사유로 안전하게 거부하지만, 현재는 RPC 응답과 `intent_log` 외에 운영자에게 즉시 전달되는 신호가 없다. Strategy가 마지막 slots RPC 이후 발생한 warning을 알지 못한 채 intent를 제출하는 race에서도 거부 자체는 정상 동작하지만, 운영자는 반복되는 거부와 전략 정체의 원인을 별도 조사 전까지 인지하기 어렵다.
|
||||
|
||||
기대 상태는 durable intent idempotency와 기존 reconciliation 차단 권한을 유지하면서, 새 `intent_id`가 최초로 `reconcile_warned` 최종 거부된 시점에만 allowlist 기반 `trex.alerts` 경고를 best-effort로 한 건 발행하는 것이다. 동일 `intent_id`의 retry는 저장된 결정을 그대로 반환하되 추가 경고를 만들지 않아야 하며, 알림 실패가 intent 거부·응답·저장 또는 다른 business path를 바꾸어서는 안 된다.
|
||||
|
||||
## Evidence
|
||||
|
||||
- GitHub issue [#74](https://github.com/fourblocks2020/trex-core/issues/74)는 실운영에서 warning 활성 뒤 strategy intent가 거부됐지만 즉시 알림이 없어 원인 파악이 지연된 사례를 기록하고, 최초 `reconcile_warned` 거부 알림·중복 억제·민감정보 제외·non-blocking 동작을 요구한다. 확인 시 issue는 open이고 후속 comment나 연결된 PR은 없었다.
|
||||
- 현재 `trex_core/application/intents.py`의 `submit_buy()`와 `submit_sell()`은 `IntentStore.try_insert()`를 dedup gate로 호출한 뒤 admission 결과를 `finalize()`하고 반환한다. Duplicate `intent_id`는 `_dedup()`에서 기존 결정을 반환하므로 새 admission을 실행하지 않는다.
|
||||
- 같은 파일의 buy·sell warning gate는 `warn_store.is_warned(slot_id)`가 참이면 `IntentDecision(status="rejected", reason="reconcile_warned")`을 반환한다. `IntentService` constructor에는 `alarm_emitter` 의존성이 없고, 이 최종 거부를 `trex.alerts`에 발행하는 호출도 없다.
|
||||
- `trex_core/runtime/daemon.py`는 하나의 `AlarmEmitter`를 생성해 ledger, kill-switch, scheduler와 execution service에 주입하지만 `IntentService`에는 주입하지 않는다. 따라서 production intent path에는 재사용 가능한 publisher가 존재하지만 wiring이 빠져 있다.
|
||||
- `trex_core/application/alarm_emitter.py`의 `publish_alarm()`은 `severity`로 `critical | warn | info`를 사용하고 publish exception을 삼켜 호출자의 business path를 중단하지 않는 best-effort 계약이다. `docs/interface-contract.md`도 warning wire literal을 `warn`으로 정의한다.
|
||||
- `trex_core/adapters/persistence/postgresql.py`의 `PostgreSQLIntentStore`는 `intent_id` unique insert를 durable dedup gate로 사용하고 final `status`, `slot_id`, `reason`을 저장한다. 별도 notification state 없이도 “최초 처리된 intent만 발행 후보”라는 경계는 이 기존 gate로 정할 수 있다.
|
||||
- `tests/application/test_reconcile_warning_blocking.py`와 `tests/application/test_slots_handler_quantity_mismatch.py`는 buy/sell `reconcile_warned` 거부와 last-RPC race를 검증하지만 alarm 발행을 단언하지 않는다. `tests/application/test_messaging_intent_consumer.py`는 duplicate `intent_id`가 저장된 동일 결정을 반환하는 idempotency를 검증한다.
|
||||
- 현재 `main` revision `2e314232e3073381b6fc9ee268ae8e12fbcc8749`에서 관련 baseline인 `tests/application/test_reconcile_warning_blocking.py`, `tests/application/test_slots_handler_quantity_mismatch.py`, `tests/application/test_messaging_intent_consumer.py`, `tests/application/test_alarm_emitter.py`를 실행한 결과 `84 passed`였다.
|
||||
- Git history에서 `c193388`은 external strategy buy/sell에 warning gate를 의도적으로 추가하면서 safety-sell·kill-switch·watchdog가 `IntentService`를 우회하도록 정했다. 이번 Line은 그 차단 정책이나 내부 안전 경로를 변경하지 않고, 이미 확정된 외부 intent 거부의 운영 가시성만 추가한다.
|
||||
|
||||
## Scope
|
||||
|
||||
- AC admission은 새 `AC-0025` **create**로 분류한다. 가장 가까운 `AC-0023`은 C-case strategy RPC 제외·intent blocking·warning clear 권한을 소유하고, `AC-0024`는 B/C mismatch 자체의 detected/changed/resolved lifecycle을 소유한다. 이번 동작은 “외부 strategy intent가 실제로 차단된 최초 사건”이라는 별도 운영 신호이며 독립 PASS/FAIL과 장기 active 가치가 있으므로 기존 AC의 update 또는 satisfy로 흡수하지 않는다. `update`, `retire`, `satisfy`, `release evidence` 대상은 없다.
|
||||
- `IntentService`의 buy·sell 처리에서 새 `intent_id`의 최종 decision이 `status="rejected"` 및 `reason="reconcile_warned"`로 확정되고 `IntentStore.finalize()`가 성공한 뒤에만 `AlarmEmitter.publish_alarm()`을 한 번 호출한다.
|
||||
- Alarm은 기존 wire 계약에 맞춰 `severity="warn"`, code `strategy_intent_rejected`를 사용한다. Fields는 최소 allowlist인 `instance_key`, `intent_id`, `intent_type`, `reason`, `broker_submitted=false`를 포함하고, 요청 또는 canonical slot에서 안전하게 확인 가능한 경우에만 `slot_id`, `portfolio_id`, `symbol`, normalized `strategy_id`를 추가한다.
|
||||
- Buy의 canonical `slot_id` 입력 또는 legacy `portfolio_id` mapping과 sell의 `slot_id` 입력 모두에서 가능한 slot context를 수집하되, context 조회 실패나 optional field 부재는 원래 `reconcile_warned` decision을 바꾸거나 알림을 위해 새로운 mutation을 유발하지 않는다.
|
||||
- 동일 `intent_id`의 duplicate/retry는 기존 durable `intent_log` 결정을 반환하고 alarm을 재발행하지 않는다. 최초 처리 중 관찰되는 `processing`/`duplicate_in_flight` 응답도 별도 alarm 대상이 아니다.
|
||||
- Alarm publish 성공 여부와 무관하게 원래 거부 decision과 RPC 응답은 유지된다. Publisher 부재와 transport failure는 non-blocking이며 별도 durable outbox, notification-state row 또는 retry loop를 만들지 않는다.
|
||||
- Alarm title·fields·rendered log에는 credential, token, account number/hash, DSN, raw strategy payload, raw broker/provider payload와 raw exception body를 포함하지 않는다.
|
||||
- Production daemon은 이미 생성하는 동일 `AlarmEmitter`를 `IntentService`에 주입한다. Optional dependency를 사용하는 직접 application test와 non-daemon construction은 기존 동작을 유지한다.
|
||||
- Verification은 buy와 sell의 최초 경고 각 한 건, 같은 `intent_id` retry의 추가 경고 0건, 다른 `intent_id`의 독립 경고, admitted 및 다른 rejection reason의 무발행, publish failure의 non-blocking, payload allowlist·`broker_submitted=false`, production wiring과 기존 warning blocking/idempotency regression을 포함한다.
|
||||
|
||||
## Out of Scope
|
||||
|
||||
- `AC-0023`의 B/C warning 판정, slots RPC 제외, buy/sell 차단, fail-closed 및 authoritative clear semantics 변경.
|
||||
- `AC-0024`의 mismatch detected/changed/resolved alarm과 `reconcile_notification_state` 변경 또는 통합.
|
||||
- Strategy retry·backoff·pause·circuit-breaker 정책, strategy 또는 TVR wire schema 변경, 새로운 broker order submit/cancel 동작.
|
||||
- Safety-sell, kill-switch, watchdog 등 `IntentService`를 우회하는 내부 안전 경로의 차단 또는 알림.
|
||||
- 다른 rejection reason에 대한 범용 intent rejection alarm, 알림 집계·rate limit·severity escalation.
|
||||
- Alarm delivery를 위한 durable outbox, 별도 notification table, 재시작 후 재발행 또는 publish retry 보장. `finalize()` 성공과 alarm publish 사이 process crash로 인한 알림 유실 가능성은 이 Line에서 해소하지 않는다.
|
||||
- `trex-notifier`의 message formatting·routing·delivery channel 변경과 운영 배포 자체.
|
||||
|
||||
## Risks and Unknowns
|
||||
|
||||
### Risks
|
||||
|
||||
- `IntentStore.finalize()` 뒤 best-effort publish를 수행하므로 process가 그 사이 종료되거나 transport가 실패하면 durable decision은 남지만 alarm은 유실된다. Issue가 outbox/retry를 제외했으므로 이 Line은 business path 독립성을 우선하며, delivery 보장이 필요해지면 별도 Requirement로 다룬다.
|
||||
- Buy는 canonical `slot_id`와 legacy `portfolio_id` 두 입력을 허용하고 sell은 `sell_strategy_id`를 사용한다. Context 정규화가 admission logic을 복제하면 drift 위험이 있으므로, 알림은 decision에 영향을 주지 않는 helper와 allowlist output으로 제한하고 양쪽 입력 형태를 테스트해야 한다.
|
||||
- 기존 `AlarmEmitter`는 여러 subsystem이 공유한다. Intent-specific 실패 처리나 logging 변경이 공용 emitter 계약을 확장하지 않도록 production wiring과 publisher failure test를 분리해야 한다.
|
||||
- Rejection payload에 raw request 또는 `extra` dict를 복사하면 민감정보가 유출될 수 있다. 허용된 scalar field만 명시적으로 구성하고 forbidden key/value sentinel을 rendered payload와 log에서 검사해야 한다.
|
||||
|
||||
### Assumptions
|
||||
|
||||
- Issue #74의 “경고(WARNING)”는 현재 interface contract의 wire literal `severity="warn"`을 의미한다.
|
||||
- “최초”의 canonical identity는 기존 `IntentStore.try_insert()`가 소유하는 `intent_id`이며, process-local set이나 새로운 persistence를 추가하지 않는다.
|
||||
- `strategy_intent_rejected`는 issue가 제시한 이 기능의 canonical alarm code로 사용하며, notifier-side inventory 등록이 별도로 필요하더라도 core의 event contract를 변경하지 않는다.
|
||||
@@ -0,0 +1,3 @@
|
||||
---
|
||||
id: "line-0008"
|
||||
---
|
||||
@@ -0,0 +1,51 @@
|
||||
---
|
||||
id: "req-0008"
|
||||
status: approved
|
||||
discovery: "dcy-0008"
|
||||
criteria:
|
||||
create:
|
||||
- "ac-0025"
|
||||
update: []
|
||||
retire: []
|
||||
satisfy: []
|
||||
---
|
||||
|
||||
# Reconcile-warned strategy intent rejection alerts
|
||||
|
||||
## Objective
|
||||
|
||||
`trex-core`가 활성 reconciliation warning 때문에 외부 strategy의 신규 buy 또는 sell intent를 `reconcile_warned`로 최종 거부하면, 해당 `intent_id`의 최초 처리에서만 원인을 식별할 수 있는 sanitized warning event를 `trex.alerts`에 한 건 발행하게 한다. 기존 durable intent idempotency와 warning 차단 권한을 유지하고, duplicate intent 및 무관한 decision의 알림을 억제하며, 알림 실패가 intent decision·저장·응답 또는 broker 동작에 영향을 주지 않도록 한다.
|
||||
|
||||
## Scope
|
||||
|
||||
- `IntentService`가 새 `intent_id`를 처리하여 buy 또는 sell decision을 `status="rejected"`, `reason="reconcile_warned"`로 결정하고 `IntentStore.finalize()`가 성공한 뒤에만 warning event를 발행한다.
|
||||
- 동일 `intent_id`의 duplicate/retry는 기존 `IntentStore.try_insert()`와 저장된 final decision을 canonical dedup authority로 사용한다. Duplicate path, 처리 중인 intent 및 process-local cache는 새 event를 만들지 않는다.
|
||||
- 서로 다른 `intent_id`가 같은 warned slot을 대상으로 각각 거부되면 각 intent는 독립적인 최초 거부 event 한 건을 만든다.
|
||||
- Event는 기존 `AlarmEmitter`와 `trex.alerts` envelope·routing contract를 재사용하고 `severity="warn"`, `code="strategy_intent_rejected"`를 사용한다.
|
||||
- Event fields의 필수 allowlist는 `instance_key`, `intent_id`, `intent_type`, `reason="reconcile_warned"`, `broker_submitted=false`다. Canonical slot 또는 intent에서 안전하게 확인된 값이 있을 때만 `slot_id`, `portfolio_id`, `symbol`, `strategy_id`를 추가하며, unavailable optional field는 추정하거나 raw payload에서 복사하지 않고 생략한다.
|
||||
- Buy의 `strategy_id`와 sell의 `sell_strategy_id`는 event의 공통 `strategy_id` field로 정규화한다. Buy의 canonical `slot_id` 또는 legacy `portfolio_id` 입력과 sell의 `slot_id` 입력 모두에서 같은 allowlist 계약을 적용한다.
|
||||
- Optional context 조회 실패, publisher 부재 및 alarm transport failure는 원래 final decision과 RPC response를 변경하거나 intent 처리를 실패시키지 않는다. 알림을 위해 slot, warning, intent, broker, ledger 또는 portfolio-hub 상태를 mutation하지 않는다.
|
||||
- Production daemon은 이미 생성하는 동일 `AlarmEmitter`를 `IntentService`에 연결한다. 직접 application construction과 non-daemon entry point는 optional publisher 경계에서 기존 동작과 호환된다.
|
||||
- Buy·sell 최초 거부, duplicate 억제, 서로 다른 intent의 독립 발행, 다른 decision의 무발행, publish failure, optional context, payload 보안과 production wiring을 repository-local automated tests로 검증한다.
|
||||
|
||||
## Constraints
|
||||
|
||||
- AC-0023이 소유하는 B/C warning 판정, strategy RPC 제외, buy/sell `reconcile_warned` 차단, warn-store fail-closed 및 authoritative clear semantics를 완화·대체·확장하지 않는다.
|
||||
- AC-0024의 mismatch detected/changed/resolved lifecycle, fingerprint, `reconcile_notification_state`와 operator `sync`/`verify` recovery authority를 변경하거나 재사용하지 않는다.
|
||||
- Alarm은 `IntentStore.finalize()`보다 먼저 발행해서는 안 된다. Finalize가 실패하거나 final decision이 확정되지 않은 경우 알림을 성공한 거부로 보고해서는 안 된다.
|
||||
- Publisher return value나 failure는 저장된 rejection을 rollback하거나 RPC status/reason을 바꾸지 않는다. 별도 durable outbox, notification-state persistence 또는 retry loop를 추가하지 않는다.
|
||||
- Event title, fields와 이 경로가 남기는 rendered log에는 account number/hash, credential, token, password, DSN, raw strategy `extra`, raw request, raw KIS/broker/portfolio-hub payload 또는 raw exception body가 없어야 한다.
|
||||
- `broker_submitted=false`는 거부가 broker submission 전에 확정됐음을 나타내며, 이 기능이 broker submit/cancel 호출을 새로 만들거나 기존 호출 순서를 바꾸어서는 안 된다.
|
||||
- Safety-sell, kill-switch, watchdog처럼 `IntentService`를 우회하는 내부 안전 경로는 계속 reconciliation warning gate와 이 strategy rejection alarm의 대상이 아니다.
|
||||
- 기존 intent RPC response schema, intent idempotency, slot transition과 alarm wire compatibility를 유지하며 RabbitMQ topology 또는 notifier rendering 변경을 요구하지 않는다.
|
||||
- Normative Design Documents: 없음. 이 REQ와 AC-0025가 normative specification을 소유한다.
|
||||
|
||||
## Non-Goals
|
||||
|
||||
- 다른 rejection reason의 범용 strategy intent alarm, reason별 severity 정책, rate limit·aggregation·escalation
|
||||
- Strategy retry·backoff·pause·circuit-breaker 정책과 strategy/TVR wire schema 변경
|
||||
- B/C mismatch 탐지·해소 알림과 strategy intent 거부 알림의 lifecycle 또는 persistence 통합
|
||||
- Alarm의 exactly-once delivery, process crash 복구, durable outbox, publish retry와 재시작 후 재발행
|
||||
- Warning 자동 해제, quantity 자동 보정, broker order submit/cancel, ledger write 또는 portfolio-hub write
|
||||
- Safety-sell·kill-switch·watchdog 내부 경로의 신규 알림
|
||||
- `trex-notifier` message formatting·delivery channel, RabbitMQ topology, production deployment, release 또는 issue closure
|
||||
@@ -15,6 +15,7 @@ import pytest
|
||||
|
||||
from trex_core.adapters.messaging.in_memory import InMemoryTransport
|
||||
from trex_core.application.alarm_emitter import AlarmEmitter
|
||||
from trex_core.application.messaging import TransportUnavailableError
|
||||
|
||||
# ------------------------------------------------------------------ helpers
|
||||
|
||||
@@ -154,3 +155,40 @@ def test_alarm_emitter_no_telegram_reference():
|
||||
source = inspect.getsource(ae_mod)
|
||||
assert "TELEGRAM" not in source, "alarm_emitter must not reference TELEGRAM_*"
|
||||
assert "telegram" not in source.lower(), "alarm_emitter must not reference telegram"
|
||||
|
||||
|
||||
def test_publish_alarm_transport_unavailable_log_is_sanitized(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
):
|
||||
class UnavailableTransport:
|
||||
def publish_alarm_event(self, routing_key, payload):
|
||||
raise TransportUnavailableError("secret-token-value")
|
||||
|
||||
def health(self):
|
||||
from trex_core.application.messaging import TransportHealth
|
||||
return TransportHealth.DISCONNECTED
|
||||
|
||||
emitter = AlarmEmitter(UnavailableTransport()) # type: ignore[arg-type]
|
||||
with caplog.at_level(logging.ERROR, logger="trex_core.application.alarm_emitter"):
|
||||
emitter.publish_alarm("strategy_intent_rejected", "alarm title")
|
||||
|
||||
assert any("transport unavailable" in r.message for r in caplog.records)
|
||||
assert "secret-token-value" not in caplog.text
|
||||
|
||||
|
||||
def test_publish_alarm_generic_exception_log_is_sanitized(caplog: pytest.LogCaptureFixture):
|
||||
class BrokenTransport:
|
||||
def publish_alarm_event(self, routing_key, payload):
|
||||
raise RuntimeError("******")
|
||||
|
||||
def health(self):
|
||||
from trex_core.application.messaging import TransportHealth
|
||||
return TransportHealth.CONNECTED
|
||||
|
||||
emitter = AlarmEmitter(BrokenTransport()) # type: ignore[arg-type]
|
||||
with caplog.at_level(logging.ERROR, logger="trex_core.application.alarm_emitter"):
|
||||
emitter.publish_alarm("strategy_intent_rejected", "alarm title")
|
||||
|
||||
assert any("alarm publish failed" in r.message for r in caplog.records)
|
||||
assert "RuntimeError" in caplog.text
|
||||
assert "******" not in caplog.text
|
||||
|
||||
@@ -3,6 +3,8 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
import pytest
|
||||
|
||||
from trex_core.application.broker_port import BalanceItem
|
||||
from trex_core.application.intents import BuyIntent, IntentService, SellIntent
|
||||
from trex_core.application.messaging import RpcRequest
|
||||
@@ -58,6 +60,72 @@ class InMemoryIntentStore:
|
||||
self._db[intent_id] = {"status": status, "slot_id": slot_id, "reason": reason}
|
||||
|
||||
|
||||
class RecordingIntentStore(InMemoryIntentStore):
|
||||
def __init__(self, trace=None, fail_finalize=False):
|
||||
super().__init__()
|
||||
self._trace = trace
|
||||
self._fail_finalize = fail_finalize
|
||||
|
||||
def finalize(self, intent_id, status, slot_id, reason):
|
||||
if self._trace is not None:
|
||||
self._trace.append("finalize")
|
||||
if self._fail_finalize:
|
||||
raise RuntimeError("finalize failed")
|
||||
super().finalize(intent_id, status, slot_id, reason)
|
||||
|
||||
|
||||
class RecordingAlarmEmitter:
|
||||
def __init__(self, *, trace=None, return_value=True, raise_error=False):
|
||||
self.calls = []
|
||||
self._trace = trace
|
||||
self._return_value = return_value
|
||||
self._raise_error = raise_error
|
||||
|
||||
def publish_alarm(self, code, title, *, fields=None, severity="critical"):
|
||||
if self._trace is not None:
|
||||
self._trace.append("publish")
|
||||
self.calls.append((code, title, severity, fields or {}))
|
||||
if self._raise_error:
|
||||
raise RuntimeError("transport down")
|
||||
return self._return_value
|
||||
|
||||
|
||||
class DuplicateInFlightIntentStore:
|
||||
def try_insert(self, intent_id, intent_type, payload):
|
||||
return False
|
||||
|
||||
def get(self, intent_id):
|
||||
return None
|
||||
|
||||
def finalize(self, intent_id, status, slot_id, reason):
|
||||
raise AssertionError("finalize must not be called for duplicate_in_flight")
|
||||
|
||||
|
||||
class PostFinalizeContextFailingSlotRepository(InMemorySlotRepository):
|
||||
"""Raise fetch_slot only after explicitly armed by finalize."""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self._raise_on_fetch = False
|
||||
|
||||
def fetch_slot(self, slot_id):
|
||||
if self._raise_on_fetch:
|
||||
raise RuntimeError("context lookup failed")
|
||||
return super().fetch_slot(slot_id)
|
||||
|
||||
|
||||
class FinalizeArmingIntentStore(InMemoryIntentStore):
|
||||
"""Arm repo fetch failure immediately after finalize persistence."""
|
||||
|
||||
def __init__(self, repo):
|
||||
super().__init__()
|
||||
self._repo = repo
|
||||
|
||||
def finalize(self, intent_id, status, slot_id, reason):
|
||||
super().finalize(intent_id, status, slot_id, reason)
|
||||
self._repo._raise_on_fetch = True
|
||||
|
||||
|
||||
def _make_holding_slot(repo, svc, instance_key="default", portfolio_id="pf-A", symbol="005930", qty=10):
|
||||
"""Helper: creates universe, puts slot in HOLDING state with given qty."""
|
||||
repo.reconcile_universe(instance_key, {portfolio_id: symbol})
|
||||
@@ -272,6 +340,256 @@ def test_intent_service_sell_rejected_when_reconcile_warned():
|
||||
assert d.reason == "reconcile_warned"
|
||||
|
||||
|
||||
def test_reconcile_warned_buy_emits_alarm_with_allowlisted_fields_after_finalize():
|
||||
repo = InMemorySlotRepository()
|
||||
svc = SlotService(repo)
|
||||
warn_store = InMemoryReconcileWarningStore()
|
||||
trace = []
|
||||
intent_store = RecordingIntentStore(trace=trace)
|
||||
alarms = RecordingAlarmEmitter(trace=trace)
|
||||
|
||||
repo.reconcile_universe("default", {"pf-A": "005930"})
|
||||
slot = repo.find_slot_by_portfolio("default", "pf-A")
|
||||
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
|
||||
|
||||
intent_svc = IntentService(
|
||||
svc, repo, intent_store,
|
||||
instance_key="default",
|
||||
warn_store=warn_store,
|
||||
alarm_emitter=alarms,
|
||||
)
|
||||
d = intent_svc.submit_buy(BuyIntent(
|
||||
intent_id="b-alert-001",
|
||||
slot_id=slot.slot_id,
|
||||
entry_price=100.0,
|
||||
qty=10.0,
|
||||
strategy_id="stg-buy",
|
||||
))
|
||||
assert d.status == "rejected"
|
||||
assert d.reason == "reconcile_warned"
|
||||
assert trace == ["finalize", "publish"]
|
||||
assert len(alarms.calls) == 1
|
||||
code, title, severity, fields = alarms.calls[0]
|
||||
assert code == "strategy_intent_rejected"
|
||||
assert severity == "warn"
|
||||
assert title
|
||||
assert fields["instance_key"] == "default"
|
||||
assert fields["intent_id"] == "b-alert-001"
|
||||
assert fields["intent_type"] == "buy"
|
||||
assert fields["reason"] == "reconcile_warned"
|
||||
assert fields["broker_submitted"] is False
|
||||
assert fields["slot_id"] == str(slot.slot_id)
|
||||
assert fields["portfolio_id"] == "pf-A"
|
||||
assert fields["symbol"] == "005930"
|
||||
assert fields["strategy_id"] == "stg-buy"
|
||||
for blocked in ("qty", "entry_price", "limit_price", "venue", "extra", "sell_strategy_id"):
|
||||
assert blocked not in fields
|
||||
text = repr((title, fields))
|
||||
for forbidden in ("account", "token", "password", "secret", "dsn", "connection", "raw_payload"):
|
||||
assert forbidden not in text.lower()
|
||||
|
||||
|
||||
def test_reconcile_warned_sell_emits_alarm_with_normalized_strategy_id():
|
||||
repo = InMemorySlotRepository()
|
||||
svc = SlotService(repo)
|
||||
warn_store = InMemoryReconcileWarningStore()
|
||||
intent_store = InMemoryIntentStore()
|
||||
alarms = RecordingAlarmEmitter()
|
||||
|
||||
slot = _make_holding_slot(repo, svc, qty=10, portfolio_id="pf-A", symbol="005930")
|
||||
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C")
|
||||
|
||||
intent_svc = IntentService(
|
||||
svc, repo, intent_store,
|
||||
instance_key="default",
|
||||
warn_store=warn_store,
|
||||
alarm_emitter=alarms,
|
||||
)
|
||||
d = intent_svc.submit_sell(SellIntent(
|
||||
intent_id="s-alert-001",
|
||||
slot_id=slot.slot_id,
|
||||
reason="strategy_exit",
|
||||
sell_strategy_id="stg-sell",
|
||||
))
|
||||
assert d.status == "rejected"
|
||||
assert d.reason == "reconcile_warned"
|
||||
assert len(alarms.calls) == 1
|
||||
_, _, severity, fields = alarms.calls[0]
|
||||
assert severity == "warn"
|
||||
assert fields["intent_type"] == "sell"
|
||||
assert fields["strategy_id"] == "stg-sell"
|
||||
assert fields["slot_id"] == str(slot.slot_id)
|
||||
|
||||
|
||||
def test_reconcile_warned_alarm_dedup_by_intent_id_across_service_recreation():
|
||||
repo = InMemorySlotRepository()
|
||||
svc = SlotService(repo)
|
||||
warn_store = InMemoryReconcileWarningStore()
|
||||
intent_store = InMemoryIntentStore()
|
||||
alarms = RecordingAlarmEmitter()
|
||||
|
||||
repo.reconcile_universe("default", {"pf-A": "005930"})
|
||||
slot = repo.find_slot_by_portfolio("default", "pf-A")
|
||||
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
|
||||
|
||||
s1 = IntentService(
|
||||
svc, repo, intent_store, instance_key="default", warn_store=warn_store, alarm_emitter=alarms
|
||||
)
|
||||
s2 = IntentService(
|
||||
svc, repo, intent_store, instance_key="default", warn_store=warn_store, alarm_emitter=alarms
|
||||
)
|
||||
first = s1.submit_buy(BuyIntent(intent_id="b-dedup", slot_id=slot.slot_id))
|
||||
second = s1.submit_buy(BuyIntent(intent_id="b-dedup", slot_id=slot.slot_id))
|
||||
third = s2.submit_buy(BuyIntent(intent_id="b-dedup", slot_id=slot.slot_id))
|
||||
assert first.reason == "reconcile_warned"
|
||||
assert second.reason == "reconcile_warned"
|
||||
assert third.reason == "reconcile_warned"
|
||||
assert len(alarms.calls) == 1
|
||||
|
||||
|
||||
def test_reconcile_warned_alarm_emits_per_distinct_intent_id_on_same_slot():
|
||||
repo = InMemorySlotRepository()
|
||||
svc = SlotService(repo)
|
||||
warn_store = InMemoryReconcileWarningStore()
|
||||
intent_store = InMemoryIntentStore()
|
||||
alarms = RecordingAlarmEmitter()
|
||||
|
||||
repo.reconcile_universe("default", {"pf-A": "005930"})
|
||||
slot = repo.find_slot_by_portfolio("default", "pf-A")
|
||||
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
|
||||
|
||||
intent_svc = IntentService(
|
||||
svc, repo, intent_store, instance_key="default", warn_store=warn_store, alarm_emitter=alarms
|
||||
)
|
||||
d1 = intent_svc.submit_buy(BuyIntent(intent_id="b-uniq-1", slot_id=slot.slot_id))
|
||||
d2 = intent_svc.submit_buy(BuyIntent(intent_id="b-uniq-2", slot_id=slot.slot_id))
|
||||
assert d1.reason == "reconcile_warned"
|
||||
assert d2.reason == "reconcile_warned"
|
||||
assert len(alarms.calls) == 2
|
||||
|
||||
|
||||
def test_reconcile_warned_alarm_not_published_for_duplicate_in_flight_or_other_decisions():
|
||||
repo = InMemorySlotRepository()
|
||||
svc = SlotService(repo)
|
||||
alarms = RecordingAlarmEmitter()
|
||||
|
||||
duplicate = IntentService(
|
||||
svc, repo, DuplicateInFlightIntentStore(), instance_key="default", alarm_emitter=alarms
|
||||
)
|
||||
d_dup = duplicate.submit_buy(BuyIntent(intent_id="dup", portfolio_id="pf-A"))
|
||||
assert d_dup.reason == "duplicate_in_flight"
|
||||
assert alarms.calls == []
|
||||
|
||||
repo.reconcile_universe("default", {"pf-A": "005930"})
|
||||
slot = repo.find_slot_by_portfolio("default", "pf-A")
|
||||
normal = IntentService(
|
||||
svc, repo, InMemoryIntentStore(), instance_key="default", alarm_emitter=alarms
|
||||
)
|
||||
admitted = normal.submit_buy(BuyIntent(intent_id="admitted", slot_id=slot.slot_id, qty=1, entry_price=100))
|
||||
assert admitted.status == "admitted"
|
||||
invalid = normal.submit_buy(BuyIntent(intent_id="bad-venue", slot_id=slot.slot_id, venue="BAD"))
|
||||
assert invalid.reason == "invalid_venue"
|
||||
repo.reconcile_universe("default", {"pf-A": "005930", "pf-B": "000660"})
|
||||
slot_b = repo.find_slot_by_portfolio("default", "pf-B")
|
||||
capped = IntentService(
|
||||
svc, repo, InMemoryIntentStore(), instance_key="default", max_active_slots=0, alarm_emitter=alarms
|
||||
)
|
||||
full = capped.submit_buy(BuyIntent(intent_id="full", slot_id=slot_b.slot_id))
|
||||
assert full.reason == "capacity_full"
|
||||
assert alarms.calls == []
|
||||
|
||||
|
||||
def test_reconcile_warned_alarm_finalize_failure_and_publish_failure_are_isolated():
|
||||
repo = InMemorySlotRepository()
|
||||
svc = SlotService(repo)
|
||||
warn_store = InMemoryReconcileWarningStore()
|
||||
repo.reconcile_universe("default", {"pf-A": "005930"})
|
||||
slot = repo.find_slot_by_portfolio("default", "pf-A")
|
||||
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
|
||||
|
||||
alarms = RecordingAlarmEmitter()
|
||||
failing_store = RecordingIntentStore(fail_finalize=True)
|
||||
failing_finalize = IntentService(
|
||||
svc, repo, failing_store,
|
||||
instance_key="default",
|
||||
warn_store=warn_store,
|
||||
alarm_emitter=alarms,
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="finalize failed"):
|
||||
failing_finalize.submit_buy(BuyIntent(intent_id="b-finalize-fail", slot_id=slot.slot_id))
|
||||
assert alarms.calls == []
|
||||
|
||||
soft_fail_alarm = RecordingAlarmEmitter(return_value=False)
|
||||
raising_alarm = RecordingAlarmEmitter(raise_error=True)
|
||||
non_blocking_1 = IntentService(
|
||||
svc, repo, InMemoryIntentStore(), instance_key="default", warn_store=warn_store, alarm_emitter=soft_fail_alarm
|
||||
)
|
||||
non_blocking_2 = IntentService(
|
||||
svc, repo, InMemoryIntentStore(), instance_key="default", warn_store=warn_store, alarm_emitter=raising_alarm
|
||||
)
|
||||
d1 = non_blocking_1.submit_buy(BuyIntent(intent_id="b-alarm-false", slot_id=slot.slot_id))
|
||||
d2 = non_blocking_2.submit_buy(BuyIntent(intent_id="b-alarm-exc", slot_id=slot.slot_id))
|
||||
d3 = IntentService(
|
||||
svc, repo, InMemoryIntentStore(), instance_key="default", warn_store=warn_store, alarm_emitter=None
|
||||
).submit_buy(BuyIntent(intent_id="b-no-emitter", slot_id=slot.slot_id))
|
||||
assert d1.reason == "reconcile_warned"
|
||||
assert d2.reason == "reconcile_warned"
|
||||
assert d3.reason == "reconcile_warned"
|
||||
|
||||
|
||||
def test_reconcile_warned_buy_context_lookup_failure_does_not_change_response():
|
||||
repo = PostFinalizeContextFailingSlotRepository()
|
||||
svc = SlotService(repo)
|
||||
warn_store = InMemoryReconcileWarningStore()
|
||||
intent_store = FinalizeArmingIntentStore(repo)
|
||||
alarms = RecordingAlarmEmitter()
|
||||
repo.reconcile_universe("default", {"pf-A": "005930"})
|
||||
slot = repo.find_slot_by_portfolio("default", "pf-A")
|
||||
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="B")
|
||||
intent_svc = IntentService(
|
||||
svc, repo, intent_store, instance_key="default", warn_store=warn_store, alarm_emitter=alarms
|
||||
)
|
||||
|
||||
decision = intent_svc.submit_buy(
|
||||
BuyIntent(intent_id="buy-context-fail", slot_id=slot.slot_id, strategy_id="stg-buy")
|
||||
)
|
||||
assert decision.status == "rejected"
|
||||
assert decision.reason == "reconcile_warned"
|
||||
assert len(alarms.calls) == 1
|
||||
_, _, _, fields = alarms.calls[0]
|
||||
assert fields["intent_id"] == "buy-context-fail"
|
||||
assert fields["strategy_id"] == "stg-buy"
|
||||
assert "slot_id" not in fields
|
||||
assert "portfolio_id" not in fields
|
||||
assert "symbol" not in fields
|
||||
|
||||
|
||||
def test_reconcile_warned_sell_context_lookup_failure_does_not_change_response():
|
||||
repo = PostFinalizeContextFailingSlotRepository()
|
||||
svc = SlotService(repo)
|
||||
warn_store = InMemoryReconcileWarningStore()
|
||||
intent_store = FinalizeArmingIntentStore(repo)
|
||||
alarms = RecordingAlarmEmitter()
|
||||
slot = _make_holding_slot(repo, svc, qty=10, portfolio_id="pf-A", symbol="005930")
|
||||
warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C")
|
||||
intent_svc = IntentService(
|
||||
svc, repo, intent_store, instance_key="default", warn_store=warn_store, alarm_emitter=alarms
|
||||
)
|
||||
|
||||
decision = intent_svc.submit_sell(
|
||||
SellIntent(intent_id="sell-context-fail", slot_id=slot.slot_id, sell_strategy_id="stg-sell")
|
||||
)
|
||||
assert decision.status == "rejected"
|
||||
assert decision.reason == "reconcile_warned"
|
||||
assert len(alarms.calls) == 1
|
||||
_, _, _, fields = alarms.calls[0]
|
||||
assert fields["intent_id"] == "sell-context-fail"
|
||||
assert fields["strategy_id"] == "stg-sell"
|
||||
assert "slot_id" not in fields
|
||||
assert "portfolio_id" not in fields
|
||||
assert "symbol" not in fields
|
||||
|
||||
|
||||
# REQ-104 AC-3: no warn, no pause => buy/sell admitted normally
|
||||
def test_intent_service_no_warn_admitted_normally():
|
||||
repo = InMemorySlotRepository()
|
||||
|
||||
@@ -7,6 +7,9 @@ daemon wiring was never run until the REL-004 live smoke timed out every buy.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
from trex_core.application.portfolio import PortfolioItem, SnapshotStore
|
||||
from trex_core.runtime.daemon import Daemon
|
||||
|
||||
@@ -35,3 +38,38 @@ def test_resolve_symbol_reads_item_symbol():
|
||||
def test_resolve_symbol_unknown_returns_none():
|
||||
holder = _StoreHolder(_store())
|
||||
assert Daemon._resolve_symbol(holder, "pf-missing") is None
|
||||
|
||||
|
||||
def test_daemon_wires_shared_alarm_emitter_into_intent_service():
|
||||
src = (
|
||||
Path(__file__).resolve().parents[2]
|
||||
/ "trex_core"
|
||||
/ "runtime"
|
||||
/ "daemon.py"
|
||||
).read_text(encoding="utf-8")
|
||||
tree = ast.parse(src)
|
||||
found = False
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.Assign):
|
||||
continue
|
||||
if not any(isinstance(t, ast.Attribute) and t.attr == "intent_service" for t in node.targets):
|
||||
continue
|
||||
call = node.value
|
||||
if not (
|
||||
isinstance(call, ast.Call)
|
||||
and isinstance(call.func, ast.Name)
|
||||
and call.func.id == "IntentService"
|
||||
):
|
||||
continue
|
||||
for kw in call.keywords:
|
||||
if kw.arg != "alarm_emitter":
|
||||
continue
|
||||
if (
|
||||
isinstance(kw.value, ast.Attribute)
|
||||
and kw.value.attr == "alarm_emitter"
|
||||
and isinstance(kw.value.value, ast.Name)
|
||||
and kw.value.value.id == "self"
|
||||
):
|
||||
found = True
|
||||
break
|
||||
assert found, "Daemon must pass self.alarm_emitter into IntentService(alarm_emitter=...)"
|
||||
|
||||
@@ -84,15 +84,15 @@ class AlarmEmitter:
|
||||
try:
|
||||
self._transport.publish_alarm_event(routing_key, payload)
|
||||
return True
|
||||
except TransportUnavailableError as exc:
|
||||
except TransportUnavailableError:
|
||||
_log.error(
|
||||
"alarm publish skipped (transport unavailable): code=%s: %s",
|
||||
code, exc,
|
||||
"alarm publish skipped (transport unavailable): code=%s title=%s",
|
||||
code, title,
|
||||
)
|
||||
return False
|
||||
except Exception: # noqa: BLE001 — alarm must never crash the caller
|
||||
except Exception as exc: # noqa: BLE001 — alarm must never crash the caller
|
||||
_log.error(
|
||||
"alarm publish failed: code=%s title=%s",
|
||||
code, title, exc_info=True,
|
||||
"alarm publish failed: code=%s title=%s error_type=%s",
|
||||
code, title, type(exc).__name__,
|
||||
)
|
||||
return False
|
||||
|
||||
@@ -28,6 +28,7 @@ from decimal import Decimal
|
||||
from typing import Any, Callable, Optional
|
||||
from uuid import UUID
|
||||
|
||||
from trex_core.application.alarm_emitter import AlarmEmitter
|
||||
from trex_core.application.ports import IntentStore
|
||||
from trex_core.domain.slot import (
|
||||
SlotAdmissionRejectedError,
|
||||
@@ -115,6 +116,7 @@ class IntentService:
|
||||
max_active_slots: Optional[int] = None,
|
||||
default_order_timeout_sec: Optional[int] = None,
|
||||
warn_store=None, # Optional[ReconcileWarningStore] — BAT-050 / REQ-104
|
||||
alarm_emitter: Optional[AlarmEmitter] = None,
|
||||
entry_allowed_fn: Optional[Callable[[], bool]] = None,
|
||||
buy_guard_fn: Optional[Callable[[BuyIntent], Optional[str]]] = None,
|
||||
) -> None:
|
||||
@@ -141,6 +143,8 @@ class IntentService:
|
||||
# BAT-050 / REQ-104: reconcile-warning gate store (Optional).
|
||||
# When None, gate is skipped (backward compat, NFR-4).
|
||||
self._warn_store = warn_store
|
||||
# Optional best-effort alarm publisher for reconcile_warned rejections.
|
||||
self._alarm_emitter = alarm_emitter
|
||||
# dcy-036 / REQ-114: day-start entry gate (Optional callable → bool).
|
||||
# When None or returns True, gate passes (backward compat / opt-in=False).
|
||||
# When returns False, BuyIntent is rejected with reason 'day_start_load_failed'.
|
||||
@@ -178,6 +182,13 @@ class IntentService:
|
||||
self._intents.finalize(
|
||||
intent.intent_id, decision.status, decision.slot_id, decision.reason
|
||||
)
|
||||
if decision.status == "rejected" and decision.reason == "reconcile_warned":
|
||||
self._publish_reconcile_warned_rejection_alarm(
|
||||
intent_id=intent.intent_id,
|
||||
intent_type="buy",
|
||||
strategy_id=intent.strategy_id,
|
||||
slot=self._resolve_buy_alarm_slot(intent),
|
||||
)
|
||||
return decision
|
||||
|
||||
def submit_sell(self, intent: SellIntent) -> IntentDecision:
|
||||
@@ -196,8 +207,69 @@ class IntentService:
|
||||
self._intents.finalize(
|
||||
intent.intent_id, decision.status, decision.slot_id, decision.reason
|
||||
)
|
||||
if decision.status == "rejected" and decision.reason == "reconcile_warned":
|
||||
self._publish_reconcile_warned_rejection_alarm(
|
||||
intent_id=intent.intent_id,
|
||||
intent_type="sell",
|
||||
strategy_id=intent.sell_strategy_id,
|
||||
slot=self._resolve_sell_alarm_slot(intent),
|
||||
)
|
||||
return decision
|
||||
|
||||
def _resolve_buy_alarm_slot(self, intent: BuyIntent) -> Any:
|
||||
try:
|
||||
if intent.slot_id is not None:
|
||||
return self._repo.fetch_slot(intent.slot_id)
|
||||
if intent.portfolio_id:
|
||||
return self._repo.find_slot_by_portfolio(
|
||||
self._instance_key, intent.portfolio_id
|
||||
)
|
||||
except Exception: # noqa: BLE001 - optional alarm context lookup
|
||||
return None
|
||||
return None
|
||||
|
||||
def _resolve_sell_alarm_slot(self, intent: SellIntent) -> Any:
|
||||
try:
|
||||
return self._repo.fetch_slot(intent.slot_id)
|
||||
except Exception: # noqa: BLE001 - optional alarm context lookup
|
||||
return None
|
||||
|
||||
def _publish_reconcile_warned_rejection_alarm(
|
||||
self,
|
||||
*,
|
||||
intent_id: str,
|
||||
intent_type: str,
|
||||
strategy_id: Optional[str],
|
||||
slot: Any,
|
||||
) -> None:
|
||||
if self._alarm_emitter is None:
|
||||
return
|
||||
fields: dict[str, Any] = {
|
||||
"instance_key": self._instance_key,
|
||||
"intent_id": intent_id,
|
||||
"intent_type": intent_type,
|
||||
"reason": "reconcile_warned",
|
||||
"broker_submitted": False,
|
||||
}
|
||||
if slot is not None:
|
||||
if getattr(slot, "slot_id", None) is not None:
|
||||
fields["slot_id"] = str(slot.slot_id)
|
||||
if getattr(slot, "portfolio_id", None):
|
||||
fields["portfolio_id"] = slot.portfolio_id
|
||||
if getattr(slot, "symbol", None):
|
||||
fields["symbol"] = slot.symbol
|
||||
if strategy_id:
|
||||
fields["strategy_id"] = strategy_id
|
||||
try:
|
||||
self._alarm_emitter.publish_alarm(
|
||||
"strategy_intent_rejected",
|
||||
"Strategy intent rejected by reconcile warning gate",
|
||||
fields=fields,
|
||||
severity="warn",
|
||||
)
|
||||
except Exception: # noqa: BLE001 - alarm path must not affect intent decision
|
||||
pass
|
||||
|
||||
# ------------------------------------------------------------
|
||||
def _dedup(self, intent_id: str, intent_type: str, payload: dict) -> Optional[IntentDecision]:
|
||||
"""Return prior decision if intent_id already seen; else None (and reserve it)."""
|
||||
|
||||
@@ -299,6 +299,7 @@ class Daemon:
|
||||
max_active_slots=config.max_active_slots, # dcy-020: asset-centric gate
|
||||
default_order_timeout_sec=config.default_order_timeout_sec, # dcy-022
|
||||
warn_store=self.warn_store, # BAT-050 / REQ-104: reconcile-warning gate
|
||||
alarm_emitter=self.alarm_emitter, # req-0008 / ac-0025
|
||||
entry_allowed_fn=entry_allowed_fn, # dcy-036 / REQ-114: day-start gate
|
||||
buy_guard_fn=(
|
||||
self._account_snapshot_buy_guard
|
||||
|
||||
Reference in New Issue
Block a user