Harden signal state and training maintenance workflows
This commit is contained in:
@@ -0,0 +1,242 @@
|
||||
from pathlib import Path
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.exc import SQLAlchemyError
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from src.db.engine import engine
|
||||
from src.services.maintenance.historical_compatibility import (
|
||||
build_historical_compatibility_summary,
|
||||
plan_fvg_meta_repairs,
|
||||
plan_mss_meta_repairs,
|
||||
plan_trade_signal_invalidation_repairs,
|
||||
summarize_liquidity_duplicate_groups,
|
||||
)
|
||||
|
||||
|
||||
def run_historical_compatibility_repair(
|
||||
*,
|
||||
apply_signal_invalidations: bool = False,
|
||||
apply_mss_meta: bool = False,
|
||||
apply_fvg_meta: bool = False,
|
||||
) -> dict:
|
||||
dry_run = not any((apply_signal_invalidations, apply_mss_meta, apply_fvg_meta))
|
||||
with engine.begin() as connection:
|
||||
signal_rows = _fetch_trade_signal_rows(connection)
|
||||
mss_rows = _fetch_mss_rows(connection)
|
||||
fvg_rows = _fetch_fvg_rows(connection)
|
||||
liquidity_duplicate_rows = _fetch_liquidity_duplicate_groups(connection)
|
||||
|
||||
signal_repairs = plan_trade_signal_invalidation_repairs(signal_rows)
|
||||
mss_repairs = plan_mss_meta_repairs(mss_rows)
|
||||
fvg_repairs = plan_fvg_meta_repairs(fvg_rows)
|
||||
liquidity_duplicates = summarize_liquidity_duplicate_groups(liquidity_duplicate_rows)
|
||||
|
||||
applied_repairs = {
|
||||
"signal_invalidations": 0,
|
||||
"mss_meta": 0,
|
||||
"fvg_meta": 0,
|
||||
}
|
||||
|
||||
if apply_signal_invalidations:
|
||||
for repair in signal_repairs:
|
||||
_update_trade_signal_invalidations(connection, repair)
|
||||
applied_repairs["signal_invalidations"] = len(signal_repairs)
|
||||
if apply_mss_meta:
|
||||
for repair in mss_repairs:
|
||||
_update_mss_meta(connection, repair)
|
||||
applied_repairs["mss_meta"] = len(mss_repairs)
|
||||
if apply_fvg_meta:
|
||||
for repair in fvg_repairs:
|
||||
_update_fvg_meta(connection, repair)
|
||||
applied_repairs["fvg_meta"] = len(fvg_repairs)
|
||||
|
||||
return build_historical_compatibility_summary(
|
||||
signal_repairs=signal_repairs,
|
||||
mss_repairs=mss_repairs,
|
||||
fvg_repairs=fvg_repairs,
|
||||
liquidity_duplicates=liquidity_duplicates,
|
||||
applied_repairs=applied_repairs,
|
||||
dry_run=dry_run,
|
||||
)
|
||||
|
||||
|
||||
def _fetch_trade_signal_rows(connection) -> list[dict]:
|
||||
sql = text(
|
||||
"""
|
||||
select id, invalidations
|
||||
from trade_signals
|
||||
order by id asc
|
||||
"""
|
||||
)
|
||||
return [dict(row) for row in connection.execute(sql).mappings().all()]
|
||||
|
||||
|
||||
def _fetch_mss_rows(connection) -> list[dict]:
|
||||
sql = text(
|
||||
"""
|
||||
select id, accepted, meta
|
||||
from structure_events
|
||||
where event_type = 'MSS'
|
||||
order by id asc
|
||||
"""
|
||||
)
|
||||
return [dict(row) for row in connection.execute(sql).mappings().all()]
|
||||
|
||||
|
||||
def _fetch_fvg_rows(connection) -> list[dict]:
|
||||
sql = text(
|
||||
"""
|
||||
select id, status, meta
|
||||
from fvg_zones
|
||||
order by id asc
|
||||
"""
|
||||
)
|
||||
return [dict(row) for row in connection.execute(sql).mappings().all()]
|
||||
|
||||
|
||||
def _fetch_liquidity_duplicate_groups(connection) -> list[dict]:
|
||||
sql = text(
|
||||
"""
|
||||
select
|
||||
instrument_id,
|
||||
timeframe,
|
||||
side,
|
||||
price_low,
|
||||
price_high,
|
||||
count(*) as duplicate_count,
|
||||
array_agg(id order by created_ts desc, id desc) as ids
|
||||
from liquidity_pools
|
||||
where status = 'untouched'
|
||||
group by instrument_id, timeframe, side, price_low, price_high
|
||||
having count(*) > 1
|
||||
order by count(*) desc, instrument_id asc, timeframe asc, side asc
|
||||
"""
|
||||
)
|
||||
return [dict(row) for row in connection.execute(sql).mappings().all()]
|
||||
|
||||
|
||||
def _update_trade_signal_invalidations(connection, repair: dict) -> None:
|
||||
sql = text(
|
||||
"""
|
||||
update trade_signals
|
||||
set invalidations = cast(:invalidations as jsonb)
|
||||
where id = :id
|
||||
"""
|
||||
)
|
||||
connection.execute(
|
||||
sql,
|
||||
{
|
||||
"id": int(repair["id"]),
|
||||
"invalidations": json.dumps(repair["invalidations"]),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _update_mss_meta(connection, repair: dict) -> None:
|
||||
sql = text(
|
||||
"""
|
||||
update structure_events
|
||||
set meta = cast(:meta as jsonb)
|
||||
where id = :id
|
||||
"""
|
||||
)
|
||||
connection.execute(
|
||||
sql,
|
||||
{
|
||||
"id": int(repair["id"]),
|
||||
"meta": json.dumps(repair["meta"], default=str),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _update_fvg_meta(connection, repair: dict) -> None:
|
||||
sql = text(
|
||||
"""
|
||||
update fvg_zones
|
||||
set meta = cast(:meta as jsonb)
|
||||
where id = :id
|
||||
"""
|
||||
)
|
||||
connection.execute(
|
||||
sql,
|
||||
{
|
||||
"id": int(repair["id"]),
|
||||
"meta": json.dumps(repair["meta"], default=str),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def main(argv=None, env=None) -> int:
|
||||
env = os.environ if env is None else env
|
||||
parser = _build_parser()
|
||||
args = parser.parse_args(argv)
|
||||
apply_safe_repairs = bool(args.apply_safe_repairs)
|
||||
apply_signal_invalidations = bool(args.apply_signal_invalidations or apply_safe_repairs)
|
||||
apply_mss_meta = bool(args.apply_mss_meta or apply_safe_repairs)
|
||||
apply_fvg_meta = bool(args.apply_fvg_meta or apply_safe_repairs)
|
||||
|
||||
try:
|
||||
summary = run_historical_compatibility_repair(
|
||||
apply_signal_invalidations=apply_signal_invalidations,
|
||||
apply_mss_meta=apply_mss_meta,
|
||||
apply_fvg_meta=apply_fvg_meta,
|
||||
)
|
||||
except SQLAlchemyError as exc:
|
||||
summary = {
|
||||
"passed": False,
|
||||
"dry_run": not any((apply_signal_invalidations, apply_mss_meta, apply_fvg_meta)),
|
||||
"error_type": exc.__class__.__name__,
|
||||
"error": str(exc),
|
||||
}
|
||||
_print_summary(summary, json_output=_get_bool(env, "HISTORICAL_REPAIR_OUTPUT_JSON", False))
|
||||
return 2
|
||||
|
||||
_print_summary(summary, json_output=_get_bool(env, "HISTORICAL_REPAIR_OUTPUT_JSON", False))
|
||||
return 0
|
||||
|
||||
|
||||
def _build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="scripts/run_historical_compatibility_repair.py",
|
||||
description="Scan historical compatibility issues and optionally apply safe normalization repairs.",
|
||||
)
|
||||
parser.add_argument("--apply-safe-repairs", action="store_true", help="Apply signal invalidation, MSS meta, and FVG meta repairs.")
|
||||
parser.add_argument("--apply-signal-invalidations", action="store_true", help="Apply only trade_signals invalidation normalization.")
|
||||
parser.add_argument("--apply-mss-meta", action="store_true", help="Apply only MSS meta backfill for state/status_reason.")
|
||||
parser.add_argument("--apply-fvg-meta", action="store_true", help="Apply only FVG meta backfill for status_reason/default compatibility fields.")
|
||||
return parser
|
||||
|
||||
|
||||
def _print_summary(summary: dict, json_output: bool) -> None:
|
||||
if json_output:
|
||||
print(json.dumps(summary, ensure_ascii=False, sort_keys=True, default=str))
|
||||
return
|
||||
print(f"Historical compatibility repair passed: {summary['passed']}")
|
||||
print(f"Dry run: {summary.get('dry_run')}")
|
||||
if not summary["passed"]:
|
||||
print(f"Database unavailable for historical repair: {summary.get('error_type')}")
|
||||
return
|
||||
candidate_repairs = summary.get("candidate_repairs") or {}
|
||||
print(f"Signal invalidation repairs: {candidate_repairs.get('signal_invalidations', 0)}")
|
||||
print(f"MSS meta repairs: {candidate_repairs.get('mss_meta', 0)}")
|
||||
print(f"FVG meta repairs: {candidate_repairs.get('fvg_meta', 0)}")
|
||||
print(f"Liquidity duplicate groups: {candidate_repairs.get('liquidity_duplicate_groups', 0)}")
|
||||
|
||||
|
||||
def _get_bool(env, key: str, default: bool) -> bool:
|
||||
value = env.get(key)
|
||||
if value is None or value == "":
|
||||
return default
|
||||
return str(value).strip().lower() in {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,129 @@
|
||||
from pathlib import Path
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from typing import Optional
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from src.services.training import generate_public_training_case_drafts
|
||||
|
||||
|
||||
DEFAULT_FIXTURE = ROOT / "tests" / "fixtures" / "training_case_generator_seed_v1.json"
|
||||
|
||||
|
||||
class _PublicCliError(Exception):
|
||||
def __init__(self, code: str) -> None:
|
||||
super().__init__(code)
|
||||
self.code = code
|
||||
|
||||
|
||||
def run_generator(input_path: Path, output_path: Optional[Path] = None) -> dict:
|
||||
payload = _load_payload(input_path)
|
||||
generated = generate_public_training_case_drafts(payload)
|
||||
|
||||
if output_path is not None:
|
||||
_write_output(output_path, generated)
|
||||
|
||||
event_types = sorted(
|
||||
{
|
||||
item.get("scenario", {}).get("source_event_type")
|
||||
for item in generated.get("items", [])
|
||||
if isinstance(item, dict)
|
||||
}
|
||||
)
|
||||
return {
|
||||
"passed": True,
|
||||
"input": _display_path(input_path),
|
||||
"output": _display_path(output_path) if output_path is not None else None,
|
||||
"generated": len(generated.get("items", [])),
|
||||
"event_types": event_types,
|
||||
}
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
parser = _build_parser()
|
||||
args = parser.parse_args(argv)
|
||||
input_path = Path(args.input)
|
||||
output_path = Path(args.output) if args.output else None
|
||||
try:
|
||||
summary = run_generator(input_path=input_path, output_path=output_path)
|
||||
except (_PublicCliError, OSError, ValueError, json.JSONDecodeError) as exc:
|
||||
print(json.dumps(_error_summary(input_path, output_path, exc), ensure_ascii=False))
|
||||
return 2
|
||||
|
||||
print(json.dumps(summary, ensure_ascii=False))
|
||||
return 0
|
||||
|
||||
|
||||
def _build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="scripts/run_training_case_generator.py",
|
||||
description="Offline prototype: generate public Training case drafts from trading-derived seed events.",
|
||||
epilog=(
|
||||
"Reads a local seed JSON package and writes a public draft package when --output is set. "
|
||||
"The prototype is read-only and does not write to database-backed Training tables."
|
||||
),
|
||||
)
|
||||
parser.add_argument("--input", default=str(DEFAULT_FIXTURE), help=f"Seed JSON path (default: {DEFAULT_FIXTURE.relative_to(ROOT)}).")
|
||||
parser.add_argument("--output", help="Optional output path for generated public draft package JSON.")
|
||||
return parser
|
||||
|
||||
|
||||
def _load_payload(input_path: Path) -> dict:
|
||||
with input_path.open("r", encoding="utf-8") as handle:
|
||||
payload = json.load(handle)
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError("invalid_json_body")
|
||||
return payload
|
||||
|
||||
|
||||
def _write_output(output_path: Path, payload: dict) -> None:
|
||||
try:
|
||||
if output_path.exists() and output_path.is_dir():
|
||||
raise _PublicCliError("output_path_is_directory")
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_text(
|
||||
json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
except _PublicCliError:
|
||||
raise
|
||||
except OSError as exc:
|
||||
raise _PublicCliError("output_path_unwritable") from exc
|
||||
|
||||
|
||||
def _display_path(path: Optional[Path]) -> Optional[str]:
|
||||
if path is None:
|
||||
return None
|
||||
try:
|
||||
return str(path.resolve().relative_to(ROOT))
|
||||
except ValueError:
|
||||
return str(path)
|
||||
|
||||
|
||||
def _public_error(exc: Exception) -> str:
|
||||
if isinstance(exc, _PublicCliError):
|
||||
return exc.code
|
||||
if isinstance(exc, IsADirectoryError):
|
||||
return "input_path_is_directory"
|
||||
if isinstance(exc, FileNotFoundError):
|
||||
return "input_file_not_found"
|
||||
if isinstance(exc, json.JSONDecodeError):
|
||||
return "invalid_json_body"
|
||||
return str(exc)
|
||||
|
||||
|
||||
def _error_summary(input_path: Path, output_path: Optional[Path], exc: Exception) -> dict:
|
||||
return {
|
||||
"passed": False,
|
||||
"input": _display_path(input_path),
|
||||
"output": _display_path(output_path),
|
||||
"error": _public_error(exc),
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user