433 lines
15 KiB
Python
433 lines
15 KiB
Python
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,
|
|
classify_liquidity_duplicate_groups,
|
|
plan_fvg_meta_repairs,
|
|
plan_liquidity_reference_migrations,
|
|
plan_mss_meta_repairs,
|
|
plan_trade_signal_invalidation_repairs,
|
|
)
|
|
|
|
|
|
def run_historical_compatibility_repair(
|
|
*,
|
|
apply_signal_invalidations: bool = False,
|
|
apply_mss_meta: bool = False,
|
|
apply_fvg_meta: bool = False,
|
|
apply_liquidity_safe_merges: bool = False,
|
|
apply_liquidity_reference_migrations: bool = False,
|
|
) -> dict:
|
|
dry_run = not any(
|
|
(
|
|
apply_signal_invalidations,
|
|
apply_mss_meta,
|
|
apply_fvg_meta,
|
|
apply_liquidity_safe_merges,
|
|
apply_liquidity_reference_migrations,
|
|
)
|
|
)
|
|
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_rows(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 = classify_liquidity_duplicate_groups(liquidity_duplicate_rows)
|
|
liquidity_reference_migrations = plan_liquidity_reference_migrations(liquidity_duplicates)
|
|
|
|
applied_repairs = {
|
|
"signal_invalidations": 0,
|
|
"mss_meta": 0,
|
|
"fvg_meta": 0,
|
|
"liquidity_safe_merges": 0,
|
|
"liquidity_reference_migrations": 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)
|
|
if apply_liquidity_safe_merges:
|
|
applied_repairs["liquidity_safe_merges"] = _apply_liquidity_safe_merges(connection, liquidity_duplicates)
|
|
if apply_liquidity_reference_migrations:
|
|
applied_repairs["liquidity_reference_migrations"] = _apply_liquidity_reference_migrations(
|
|
connection,
|
|
liquidity_reference_migrations,
|
|
)
|
|
|
|
return build_historical_compatibility_summary(
|
|
signal_repairs=signal_repairs,
|
|
mss_repairs=mss_repairs,
|
|
fvg_repairs=fvg_repairs,
|
|
liquidity_duplicates=liquidity_duplicates,
|
|
liquidity_reference_migrations=liquidity_reference_migrations,
|
|
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_rows(connection) -> list[dict]:
|
|
sql = text(
|
|
"""
|
|
with dupes as (
|
|
select instrument_id, timeframe, side, price_low, price_high
|
|
from liquidity_pools
|
|
where status = 'untouched'
|
|
group by instrument_id, timeframe, side, price_low, price_high
|
|
having count(*) > 1
|
|
)
|
|
select
|
|
lp.instrument_id,
|
|
lp.timeframe,
|
|
lp.side,
|
|
lp.price_low,
|
|
lp.price_high,
|
|
lp.id,
|
|
lp.pool_type,
|
|
lp.reference_ids,
|
|
lp.created_ts,
|
|
lp.meta,
|
|
coalesce(
|
|
json_agg(
|
|
json_build_object(
|
|
'id', se.id,
|
|
'sweep_ts', se.sweep_ts,
|
|
'side_swept', se.side_swept,
|
|
'sweep_high', se.sweep_high,
|
|
'sweep_low', se.sweep_low,
|
|
'close_back_inside', se.close_back_inside,
|
|
'validation_level', se.validation_level,
|
|
'status', se.status,
|
|
'meta', se.meta
|
|
)
|
|
order by se.sweep_ts, se.id
|
|
) filter (where se.id is not null),
|
|
'[]'::json
|
|
) as sweep_events,
|
|
coalesce(array_agg(se.sweep_ts order by se.sweep_ts) filter (where se.id is not null), '{}') as sweep_ts_list,
|
|
count(se.id) as sweep_count
|
|
from liquidity_pools lp
|
|
left join sweep_events se on se.pool_id = lp.id
|
|
join dupes d
|
|
on d.instrument_id = lp.instrument_id
|
|
and d.timeframe = lp.timeframe
|
|
and d.side = lp.side
|
|
and d.price_low = lp.price_low
|
|
and d.price_high = lp.price_high
|
|
where lp.status = 'untouched'
|
|
group by
|
|
lp.instrument_id,
|
|
lp.timeframe,
|
|
lp.side,
|
|
lp.price_low,
|
|
lp.price_high,
|
|
lp.id,
|
|
lp.pool_type,
|
|
lp.reference_ids,
|
|
lp.created_ts,
|
|
lp.meta
|
|
order by
|
|
lp.instrument_id asc,
|
|
lp.timeframe asc,
|
|
lp.side asc,
|
|
lp.price_low asc,
|
|
lp.created_ts desc,
|
|
lp.id desc
|
|
"""
|
|
)
|
|
return [dict(row) for row in connection.execute(sql).mappings().all()]
|
|
|
|
|
|
def _apply_liquidity_safe_merges(connection, liquidity_duplicates: dict) -> int:
|
|
applied = 0
|
|
for group in liquidity_duplicates.get("groups") or []:
|
|
classification = group.get("classification")
|
|
if classification not in {"safe_auto_merge_unreferenced_superset", "safe_exact_duplicate_with_duplicate_sweeps"}:
|
|
continue
|
|
remove_pool_ids = list(group.get("remove_pool_ids") or [])
|
|
if not remove_pool_ids:
|
|
continue
|
|
if classification == "safe_exact_duplicate_with_duplicate_sweeps":
|
|
for pool_id in remove_pool_ids:
|
|
_delete_sweep_events_for_pool(connection, pool_id)
|
|
for pool_id in remove_pool_ids:
|
|
_delete_liquidity_pool(connection, pool_id)
|
|
applied += 1
|
|
return applied
|
|
|
|
|
|
def _apply_liquidity_reference_migrations(connection, liquidity_reference_migrations: dict) -> int:
|
|
applied = 0
|
|
for group in liquidity_reference_migrations.get("executable_groups") or []:
|
|
for action in group.get("actions") or []:
|
|
if action.get("action") == "rebind_sweep_event":
|
|
_rebind_sweep_event_pool(
|
|
connection,
|
|
sweep_event_id=int(action["sweep_event_id"]),
|
|
from_pool_id=int(action["from_pool_id"]),
|
|
to_pool_id=int(action["to_pool_id"]),
|
|
)
|
|
elif action.get("action") == "delete_duplicate_sweep_event":
|
|
_delete_sweep_event_by_id(
|
|
connection,
|
|
sweep_event_id=int(action["sweep_event_id"]),
|
|
pool_id=int(action["from_pool_id"]),
|
|
)
|
|
for pool_id in group.get("remove_pool_ids") or []:
|
|
_delete_liquidity_pool(connection, int(pool_id))
|
|
applied += 1
|
|
return applied
|
|
|
|
|
|
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 _delete_sweep_events_for_pool(connection, pool_id: int) -> None:
|
|
sql = text(
|
|
"""
|
|
delete from sweep_events
|
|
where pool_id = :pool_id
|
|
"""
|
|
)
|
|
connection.execute(sql, {"pool_id": int(pool_id)})
|
|
|
|
|
|
def _delete_sweep_event_by_id(connection, sweep_event_id: int, pool_id: int) -> None:
|
|
sql = text(
|
|
"""
|
|
delete from sweep_events
|
|
where id = :sweep_event_id
|
|
and pool_id = :pool_id
|
|
"""
|
|
)
|
|
connection.execute(
|
|
sql,
|
|
{
|
|
"sweep_event_id": int(sweep_event_id),
|
|
"pool_id": int(pool_id),
|
|
},
|
|
)
|
|
|
|
|
|
def _rebind_sweep_event_pool(connection, sweep_event_id: int, from_pool_id: int, to_pool_id: int) -> None:
|
|
sql = text(
|
|
"""
|
|
update sweep_events
|
|
set pool_id = :to_pool_id
|
|
where id = :sweep_event_id
|
|
and pool_id = :from_pool_id
|
|
"""
|
|
)
|
|
connection.execute(
|
|
sql,
|
|
{
|
|
"sweep_event_id": int(sweep_event_id),
|
|
"from_pool_id": int(from_pool_id),
|
|
"to_pool_id": int(to_pool_id),
|
|
},
|
|
)
|
|
|
|
|
|
def _delete_liquidity_pool(connection, pool_id: int) -> None:
|
|
sql = text(
|
|
"""
|
|
delete from liquidity_pools
|
|
where id = :pool_id
|
|
and status = 'untouched'
|
|
"""
|
|
)
|
|
connection.execute(sql, {"pool_id": int(pool_id)})
|
|
|
|
|
|
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)
|
|
apply_liquidity_safe_merges = bool(args.apply_liquidity_safe_merges or apply_safe_repairs)
|
|
apply_liquidity_reference_migrations = bool(args.apply_liquidity_reference_migrations)
|
|
|
|
try:
|
|
summary = run_historical_compatibility_repair(
|
|
apply_signal_invalidations=apply_signal_invalidations,
|
|
apply_mss_meta=apply_mss_meta,
|
|
apply_fvg_meta=apply_fvg_meta,
|
|
apply_liquidity_safe_merges=apply_liquidity_safe_merges,
|
|
apply_liquidity_reference_migrations=apply_liquidity_reference_migrations,
|
|
)
|
|
except SQLAlchemyError as exc:
|
|
summary = {
|
|
"passed": False,
|
|
"dry_run": not any(
|
|
(
|
|
apply_signal_invalidations,
|
|
apply_mss_meta,
|
|
apply_fvg_meta,
|
|
apply_liquidity_safe_merges,
|
|
apply_liquidity_reference_migrations,
|
|
)
|
|
),
|
|
"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, FVG meta, and liquidity safe merge 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.")
|
|
parser.add_argument("--apply-liquidity-safe-merges", action="store_true", help="Apply only liquidity safe merges for unreferenced superset rows and exact duplicate rows with identical sweep events.")
|
|
parser.add_argument("--apply-liquidity-reference-migrations", action="store_true", help="Apply executable liquidity reference migrations for exact duplicate pools with nonidentical sweep events, with conflict protection.")
|
|
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)}")
|
|
print(f"Liquidity safe auto-merge groups: {candidate_repairs.get('liquidity_safe_auto_merge_groups', 0)}")
|
|
print(f"Liquidity safe duplicate-sweep merge groups: {candidate_repairs.get('liquidity_safe_duplicate_sweep_merge_groups', 0)}")
|
|
print(f"Liquidity reference migration groups: {candidate_repairs.get('liquidity_reference_migration_groups', 0)}")
|
|
print(f"Liquidity reference migration executable groups: {candidate_repairs.get('liquidity_reference_migration_executable_groups', 0)}")
|
|
print(f"Liquidity reference migration blocked groups: {candidate_repairs.get('liquidity_reference_migration_blocked_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())
|