feat(execution,obs): 호가단위 스냅(dcy-030/#28) + 운영 관측성(dcy-029/#29) + REL-007 환류
dcy-030 (issue #28, P0 체결 복구): - REQ-083 KRX 호가단위(tick) 순수 유틸 (domain/pricing.py) - REQ-070 CHG: 슬리피지 실효가 호가단위 스냅 — 상한 준수(매수 floor/매도 ceil) + band<1tick 하한/상한 보장 + 부동소수 epsilon 가드. 147700→148400/54500→54700. - BAT-044 release-candidate dcy-029 (issue #29, 운영 관측성): - REQ-084 구조화 로깅 기반 (runtime/logging_setup.py): ISO8601 KST·레벨·Loki JSON ·correlation id(slot_id) - REQ-085 주문 결과 로깅 (event=order_submit|fill|reject) - REQ-087 거부 ERROR 가시성 (조용한 롤백 제거) - REQ-086 청산 reason+actor 동반 노출 (안전망/전략 구분, 신규 필드 없음) - REQ-088 운영 조회 CLI report (application/operator_report.py, 보유·오늘 체결 ·실현 PnL net·대기, direct-DB) - REQ-089 운영자 런북 docs/operations.md - REQ-090 baseline 정합 (runtime-flows·project-structure·interface-contract) - BAT-045(P0)·BAT-046(P1) release-candidate REL-007: confirmed → feedback-captured (운영 인스턴스 라이브 검증 + #28/#29 환류 정규화). BAT-037~041 released. 검증: 537 passed, 47 skipped, 회귀 0, ruff clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -395,9 +395,67 @@ def _build_parser() -> argparse.ArgumentParser:
|
||||
_add_intent_guards(p_cancel)
|
||||
_add_common(p_cancel)
|
||||
|
||||
# REQ-088 (dcy-029): operator report — holdings / today fills / realized PnL /
|
||||
# pending. Direct-DB (DB-backed point-in-time aggregate, no new RPC).
|
||||
p_report = sub.add_parser(
|
||||
"report", help="Holdings, today's fills, realized PnL, pending orders (direct-DB)"
|
||||
)
|
||||
_add_common(p_report)
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def _cmd_report(args, cfg) -> int:
|
||||
"""REQ-088: holdings / today's fills / realized PnL / pending — direct-DB."""
|
||||
import psycopg
|
||||
|
||||
from trex_core.application.operator_report import (
|
||||
Holding,
|
||||
PendingOrder,
|
||||
TradeRow,
|
||||
build_report,
|
||||
kst_day_bounds_utc,
|
||||
render_json,
|
||||
render_text,
|
||||
)
|
||||
|
||||
start_utc, end_utc = kst_day_bounds_utc()
|
||||
holdings: list[Holding] = []
|
||||
pending: list[PendingOrder] = []
|
||||
trades: list[TradeRow] = []
|
||||
with psycopg.connect(cfg.dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT symbol, portfolio_id, qty, avg_price FROM daily_slot "
|
||||
"WHERE status = 'holding' ORDER BY symbol"
|
||||
)
|
||||
for symbol, pid, qty, avg in cur.fetchall():
|
||||
holdings.append(Holding(symbol or "", pid or "",
|
||||
float(qty or 0), float(avg or 0)))
|
||||
cur.execute(
|
||||
"SELECT symbol, portfolio_id, status FROM daily_slot "
|
||||
"WHERE status IN ('pending_buy','pending_sell') ORDER BY symbol"
|
||||
)
|
||||
for symbol, pid, status in cur.fetchall():
|
||||
pending.append(PendingOrder(symbol or "", pid or "", status))
|
||||
cur.execute(
|
||||
"SELECT asset_id, side, qty, price, fee, tax FROM ledger_outbox "
|
||||
"WHERE COALESCE(executed_at, created_at) >= %s "
|
||||
"AND COALESCE(executed_at, created_at) < %s",
|
||||
(start_utc, end_utc),
|
||||
)
|
||||
for asset_id, side, qty, price, fee, tax in cur.fetchall():
|
||||
trades.append(TradeRow(asset_id, side, float(qty), float(price),
|
||||
float(fee or 0), float(tax or 0)))
|
||||
|
||||
rep = build_report(holdings, pending, trades)
|
||||
if getattr(args, "json", False):
|
||||
print(_json.dumps(render_json(rep), ensure_ascii=False))
|
||||
else:
|
||||
print(render_text(rep))
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int:
|
||||
"""CLI entrypoint.
|
||||
|
||||
@@ -417,6 +475,10 @@ def main(argv: list[str] | None = None, *, rpc=None, confirm=None) -> int:
|
||||
cfg = DaemonConfig.from_env()
|
||||
confirm = confirm or _default_confirm
|
||||
|
||||
# REQ-088 (dcy-029): operator report — direct-DB point-in-time aggregate.
|
||||
if args.cmd == "report":
|
||||
return _cmd_report(args, cfg)
|
||||
|
||||
# direct-DB control commands (pause/resume stay direct-DB, OQ-3).
|
||||
if args.cmd in ("pause", "resume", "paused"):
|
||||
store = PostgreSQLPauseStore(cfg.dsn)
|
||||
|
||||
@@ -405,7 +405,8 @@ class Daemon:
|
||||
def main() -> None: # pragma: no cover - process entrypoint
|
||||
import argparse
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
from trex_core.runtime.logging_setup import configure_logging
|
||||
configure_logging(level=logging.INFO) # REQ-084: ISO8601(KST)+level+json+cid
|
||||
parser = argparse.ArgumentParser(prog="trex-core-daemon")
|
||||
parser.add_argument(
|
||||
"--config", required=True,
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
"""구조화 로깅 기반 — ISO8601(KST) 타임스탬프 + 레벨 + Loki JSON + correlation id.
|
||||
|
||||
(BAT-045 / REQ-084, dcy-029 FR-1·FR-7)
|
||||
|
||||
기존 `logging.basicConfig(level=INFO)`는 format 미지정이라 타임스탬프·레벨 prefix가
|
||||
없었다(타임라인 재구성 불가). 이 모듈은 단일 진입점 `configure_logging()`으로:
|
||||
|
||||
- 모든 로그 라인에 ISO8601(KST, Asia/Seoul, 환경 TZ 비의존) 타임스탬프 + 레벨 부여
|
||||
- `TREX_LOG_FORMAT` env 로 `text`(기본, 사람용) / `json`(Loki 적재용) 선택
|
||||
- correlation id(intent→order→fill join key = slot_id) 를 contextvar 로 보관하고
|
||||
로깅 필터가 모든 레코드에 주입 — 한 거래의 로그를 단일 키로 묶는다
|
||||
|
||||
호출부(`_log.info(...)`)는 메시지를 바꾸지 않아도 포맷·correlation id 가 자동 적용된다.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from contextvars import ContextVar
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from typing import Iterator, Optional
|
||||
|
||||
_KST = timezone(timedelta(hours=9))
|
||||
|
||||
# 현재 처리의 correlation id (slot_id 등). 미설정이면 빈 문자열.
|
||||
_correlation_id: ContextVar[str] = ContextVar("trex_correlation_id", default="")
|
||||
|
||||
# JSON 출력에서 제외할 표준 LogRecord 속성.
|
||||
_RESERVED = {
|
||||
"name", "msg", "args", "levelname", "levelno", "pathname", "filename",
|
||||
"module", "exc_info", "exc_text", "stack_info", "lineno", "funcName",
|
||||
"created", "msecs", "relativeCreated", "thread", "threadName",
|
||||
"processName", "process", "correlation_id", "taskName",
|
||||
}
|
||||
|
||||
|
||||
def set_correlation_id(value: Optional[str]) -> None:
|
||||
"""현재 컨텍스트의 correlation id 를 설정한다(빈 값 허용)."""
|
||||
_correlation_id.set(value or "")
|
||||
|
||||
|
||||
def get_correlation_id() -> str:
|
||||
return _correlation_id.get()
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def correlation_id(value: Optional[str]) -> Iterator[None]:
|
||||
"""`with correlation_id(slot_id):` 블록 동안 correlation id 를 설정/복원한다."""
|
||||
token = _correlation_id.set(value or "")
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
_correlation_id.reset(token)
|
||||
|
||||
|
||||
class _CorrelationFilter(logging.Filter):
|
||||
"""모든 레코드에 `correlation_id` 속성을 주입한다(미설정 시 빈 문자열)."""
|
||||
|
||||
def filter(self, record: logging.LogRecord) -> bool:
|
||||
if not hasattr(record, "correlation_id"):
|
||||
record.correlation_id = _correlation_id.get()
|
||||
return True
|
||||
|
||||
|
||||
def _kst_ts(record: logging.LogRecord) -> str:
|
||||
return datetime.fromtimestamp(record.created, tz=_KST).isoformat(timespec="milliseconds")
|
||||
|
||||
|
||||
class TextFormatter(logging.Formatter):
|
||||
"""사람용 한 줄: `<ISO8601 KST> <LEVEL> <logger> [cid=<id>] <msg>`."""
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
cid = getattr(record, "correlation_id", "") or ""
|
||||
cid_part = f" [cid={cid}]" if cid else ""
|
||||
base = (
|
||||
f"{_kst_ts(record)} {record.levelname} {record.name}{cid_part} "
|
||||
f"{record.getMessage()}"
|
||||
)
|
||||
if record.exc_info:
|
||||
base += "\n" + self.formatException(record.exc_info)
|
||||
return base
|
||||
|
||||
|
||||
class JsonFormatter(logging.Formatter):
|
||||
"""Loki 적재용 구조화 한 줄 JSON."""
|
||||
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
payload = {
|
||||
"ts": _kst_ts(record),
|
||||
"level": record.levelname,
|
||||
"logger": record.name,
|
||||
"msg": record.getMessage(),
|
||||
"correlation_id": getattr(record, "correlation_id", "") or "",
|
||||
}
|
||||
# 호출부가 extra= 로 넘긴 구조화 필드를 그대로 포함.
|
||||
for k, v in record.__dict__.items():
|
||||
if k not in _RESERVED and not k.startswith("_"):
|
||||
payload.setdefault(k, v)
|
||||
if record.exc_info:
|
||||
payload["exc"] = self.formatException(record.exc_info)
|
||||
return json.dumps(payload, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
def _make_formatter() -> logging.Formatter:
|
||||
fmt = os.getenv("TREX_LOG_FORMAT", "text").strip().lower()
|
||||
return JsonFormatter() if fmt == "json" else TextFormatter()
|
||||
|
||||
|
||||
def configure_logging(level: int = logging.INFO) -> None:
|
||||
"""루트 로거를 단일 핸들러(포맷터 + correlation 필터)로 구성한다.
|
||||
|
||||
`logging.basicConfig` 를 대체한다. 재호출 시 기존 핸들러를 정리해 중복 출력을
|
||||
막는다(데몬/reset 양쪽에서 안전).
|
||||
"""
|
||||
root = logging.getLogger()
|
||||
for h in list(root.handlers):
|
||||
root.removeHandler(h)
|
||||
handler = logging.StreamHandler()
|
||||
handler.setFormatter(_make_formatter())
|
||||
handler.addFilter(_CorrelationFilter())
|
||||
root.addHandler(handler)
|
||||
root.setLevel(level)
|
||||
@@ -125,8 +125,9 @@ def main(argv: Optional[list[str]] = None) -> int: # pragma: no cover - CLI she
|
||||
import argparse
|
||||
|
||||
from trex_core.runtime.config import DaemonConfig, resolve_config_toml
|
||||
from trex_core.runtime.logging_setup import configure_logging
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
configure_logging(level=logging.INFO) # REQ-084
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="trex-core-reset",
|
||||
description="dcy-020/REQ-064: wipe the slot-core tables and reinitialize "
|
||||
|
||||
Reference in New Issue
Block a user