feat: lock trading delivery checklist and liquidity repair flow
This commit is contained in:
@@ -0,0 +1,197 @@
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
import re
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
TRADING_DELIVERY_DOCS = (
|
||||
ROOT / "README.md",
|
||||
ROOT / "RUNBOOK.md",
|
||||
)
|
||||
TRADING_DELIVERY_SECTION_HEADING = "Trading Delivery Checklist"
|
||||
TRADING_DELIVERY_COMPOSE_PREFIX = (
|
||||
"docker compose -p ai-ict -f /root/deploy/ai-ict/docker-compose.yml"
|
||||
)
|
||||
TRADING_DELIVERY_COMMAND_GROUPS = (
|
||||
(
|
||||
"pre_deploy_offline",
|
||||
(
|
||||
'.venv/bin/python -m unittest discover -s tests/unit -p "test_*.py"',
|
||||
".venv/bin/python scripts/run_offline_checks.py",
|
||||
".venv/bin/python scripts/check_local_demo_readiness.py",
|
||||
".venv/bin/python scripts/check_trading_delivery_readiness.py",
|
||||
),
|
||||
),
|
||||
(
|
||||
"database_gate",
|
||||
(
|
||||
"READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py",
|
||||
"HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py",
|
||||
".venv/bin/python scripts/run_historical_compatibility_repair.py --apply-safe-repairs",
|
||||
"HISTORICAL_REPAIR_OUTPUT_JSON=1 .venv/bin/python scripts/run_historical_compatibility_repair.py",
|
||||
"PIPELINE_OUTPUT_JSON=1 BACKTEST_REQUIRE_QUALITY_GATE=1 .venv/bin/python scripts/run_pipeline.py",
|
||||
),
|
||||
),
|
||||
(
|
||||
"deploy_and_health",
|
||||
(
|
||||
f"{TRADING_DELIVERY_COMPOSE_PREFIX} down --remove-orphans",
|
||||
f"{TRADING_DELIVERY_COMPOSE_PREFIX} up -d --build",
|
||||
f"{TRADING_DELIVERY_COMPOSE_PREFIX} ps",
|
||||
'test "$(docker ps --filter label=com.docker.compose.project=ai-ict --format \'{{.Names}}\' | wc -l | tr -d \' \')" = "1"',
|
||||
'test "$(docker ps --filter name=^/ai-ict$ --format \'{{.Names}}\' | wc -l | tr -d \' \')" = "1"',
|
||||
'test "$(docker inspect -f \'{{.State.Health.Status}}\' ai-ict)" = "healthy"',
|
||||
"curl -fsS http://127.0.0.1:8000/health",
|
||||
'curl -fsS "http://127.0.0.1:8000/signals?instrument_id=1&limit=1"',
|
||||
),
|
||||
),
|
||||
(
|
||||
"post_deploy_observe",
|
||||
(
|
||||
"READINESS_OUTPUT_JSON=1 .venv/bin/python scripts/run_database_readiness.py",
|
||||
"REPORTS_OUTPUT_JSON=1 .venv/bin/python scripts/run_reports.py",
|
||||
),
|
||||
),
|
||||
(
|
||||
"rollback_boundary",
|
||||
(
|
||||
"git rev-parse HEAD",
|
||||
f"{TRADING_DELIVERY_COMPOSE_PREFIX} logs --tail=100 ai-ict",
|
||||
"git worktree add /tmp/ai-ict-rollback <last-known-good-sha>",
|
||||
"cd /tmp/ai-ict-rollback && ./scripts/deploy_server.exp",
|
||||
"rm -rf /tmp/ai-ict-rollback",
|
||||
),
|
||||
),
|
||||
)
|
||||
TRADING_DELIVERY_COMMANDS = tuple(
|
||||
command
|
||||
for _group_name, commands in TRADING_DELIVERY_COMMAND_GROUPS
|
||||
for command in commands
|
||||
)
|
||||
TRADING_DELIVERY_REQUIRED_SNIPPETS = (
|
||||
"must-run before every market validation or deploy",
|
||||
"only when the dry-run output shows safe candidates",
|
||||
"manual review is still required for `needs_sweep_reference_migration` and `keep_version_history`",
|
||||
"paper execution boundary",
|
||||
"Rollback trigger",
|
||||
"last-known-good SHA",
|
||||
"without relying on chat memory",
|
||||
)
|
||||
TRADING_DELIVERY_REQUIRED_LOCAL_PATHS = (
|
||||
ROOT / "scripts" / "check_local_demo_readiness.py",
|
||||
ROOT / "scripts" / "deploy_server.exp",
|
||||
ROOT / "scripts" / "run_database_readiness.py",
|
||||
ROOT / "scripts" / "run_historical_compatibility_repair.py",
|
||||
ROOT / "scripts" / "run_offline_checks.py",
|
||||
ROOT / "scripts" / "run_pipeline.py",
|
||||
ROOT / "scripts" / "run_reports.py",
|
||||
)
|
||||
TRADING_DELIVERY_COMMAND_PREFIXES = (
|
||||
".venv/bin/python",
|
||||
"READINESS_OUTPUT_JSON=1",
|
||||
"HISTORICAL_REPAIR_OUTPUT_JSON=1",
|
||||
"PIPELINE_OUTPUT_JSON=1",
|
||||
"REPORTS_OUTPUT_JSON=1",
|
||||
"docker compose",
|
||||
"test ",
|
||||
"curl ",
|
||||
"git ",
|
||||
"cd /tmp/ai-ict-rollback &&",
|
||||
"rm -rf /tmp/ai-ict-rollback",
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TradingDeliveryReadinessReport:
|
||||
checked_docs: tuple[Path, ...]
|
||||
command_lines_by_doc: tuple[tuple[Path, tuple[str, ...]], ...]
|
||||
errors: tuple[str, ...]
|
||||
|
||||
|
||||
def check_trading_delivery_readiness(
|
||||
docs: tuple[Path, ...] = TRADING_DELIVERY_DOCS,
|
||||
) -> TradingDeliveryReadinessReport:
|
||||
command_lines_by_doc = []
|
||||
errors = []
|
||||
|
||||
for path in TRADING_DELIVERY_REQUIRED_LOCAL_PATHS:
|
||||
if not path.is_file():
|
||||
errors.append(f"{_display_path(path)} is missing")
|
||||
|
||||
for path in docs:
|
||||
if not path.is_file():
|
||||
command_lines_by_doc.append((path, ()))
|
||||
errors.append(f"{_display_path(path)} is missing")
|
||||
continue
|
||||
|
||||
text = path.read_text(encoding="utf-8")
|
||||
try:
|
||||
section = _trading_delivery_section(text)
|
||||
except ValueError as exc:
|
||||
command_lines_by_doc.append((path, ()))
|
||||
errors.append(f"{_display_path(path)} {exc}")
|
||||
continue
|
||||
|
||||
command_lines = _trading_delivery_command_lines(section)
|
||||
command_lines_by_doc.append((path, command_lines))
|
||||
|
||||
for snippet in TRADING_DELIVERY_REQUIRED_SNIPPETS:
|
||||
if snippet not in section:
|
||||
errors.append(f"{_display_path(path)} missing trading delivery snippet: {snippet}")
|
||||
if command_lines != TRADING_DELIVERY_COMMANDS:
|
||||
errors.append(
|
||||
f"{_display_path(path)} trading delivery commands drifted from the locked checklist shape"
|
||||
)
|
||||
|
||||
return TradingDeliveryReadinessReport(
|
||||
checked_docs=tuple(docs),
|
||||
command_lines_by_doc=tuple(command_lines_by_doc),
|
||||
errors=tuple(errors),
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
report = check_trading_delivery_readiness()
|
||||
if report.errors:
|
||||
print("Trading delivery readiness check failed")
|
||||
for error in report.errors:
|
||||
print(f"- {error}")
|
||||
return 1
|
||||
|
||||
print(
|
||||
"Trading delivery readiness check passed: "
|
||||
f"{len(report.checked_docs)} docs, "
|
||||
f"{len(TRADING_DELIVERY_COMMAND_GROUPS)} phases, "
|
||||
f"{len(TRADING_DELIVERY_COMMANDS)} locked commands."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def _trading_delivery_section(text: str) -> str:
|
||||
match = re.search(
|
||||
rf"^## {re.escape(TRADING_DELIVERY_SECTION_HEADING)}\n(?P<section>.*?)(?:\n## |\Z)",
|
||||
text,
|
||||
re.DOTALL | re.MULTILINE,
|
||||
)
|
||||
if match is None:
|
||||
raise ValueError(f"is missing the {TRADING_DELIVERY_SECTION_HEADING} section")
|
||||
return match.group("section")
|
||||
|
||||
|
||||
def _trading_delivery_command_lines(text: str) -> tuple[str, ...]:
|
||||
return tuple(
|
||||
line.strip()
|
||||
for line in text.splitlines()
|
||||
if line.strip().startswith(TRADING_DELIVERY_COMMAND_PREFIXES)
|
||||
)
|
||||
|
||||
|
||||
def _display_path(path: Path) -> str:
|
||||
try:
|
||||
return str(path.relative_to(ROOT))
|
||||
except ValueError:
|
||||
return str(path)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -14,10 +14,10 @@ if str(ROOT) not in sys.path:
|
||||
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_mss_meta_repairs,
|
||||
plan_trade_signal_invalidation_repairs,
|
||||
summarize_liquidity_duplicate_groups,
|
||||
)
|
||||
|
||||
|
||||
@@ -26,23 +26,25 @@ 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,
|
||||
) -> dict:
|
||||
dry_run = not any((apply_signal_invalidations, apply_mss_meta, apply_fvg_meta))
|
||||
dry_run = not any((apply_signal_invalidations, apply_mss_meta, apply_fvg_meta, apply_liquidity_safe_merges))
|
||||
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)
|
||||
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 = summarize_liquidity_duplicate_groups(liquidity_duplicate_rows)
|
||||
liquidity_duplicates = classify_liquidity_duplicate_groups(liquidity_duplicate_rows)
|
||||
|
||||
applied_repairs = {
|
||||
"signal_invalidations": 0,
|
||||
"mss_meta": 0,
|
||||
"fvg_meta": 0,
|
||||
"liquidity_safe_merges": 0,
|
||||
}
|
||||
|
||||
if apply_signal_invalidations:
|
||||
@@ -57,6 +59,8 @@ def run_historical_compatibility_repair(
|
||||
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)
|
||||
|
||||
return build_historical_compatibility_summary(
|
||||
signal_repairs=signal_repairs,
|
||||
@@ -102,27 +106,96 @@ def _fetch_fvg_rows(connection) -> list[dict]:
|
||||
return [dict(row) for row in connection.execute(sql).mappings().all()]
|
||||
|
||||
|
||||
def _fetch_liquidity_duplicate_groups(connection) -> list[dict]:
|
||||
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
|
||||
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
|
||||
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 _update_trade_signal_invalidations(connection, repair: dict) -> None:
|
||||
sql = text(
|
||||
"""
|
||||
@@ -174,6 +247,27 @@ def _update_fvg_meta(connection, repair: dict) -> None:
|
||||
)
|
||||
|
||||
|
||||
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_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()
|
||||
@@ -182,12 +276,14 @@ def main(argv=None, env=None) -> int:
|
||||
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)
|
||||
|
||||
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,
|
||||
)
|
||||
except SQLAlchemyError as exc:
|
||||
summary = {
|
||||
@@ -212,6 +308,7 @@ def _build_parser() -> argparse.ArgumentParser:
|
||||
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.")
|
||||
return parser
|
||||
|
||||
|
||||
@@ -229,6 +326,8 @@ def _print_summary(summary: dict, json_output: bool) -> None:
|
||||
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)}")
|
||||
|
||||
|
||||
def _get_bool(env, key: str, default: bool) -> bool:
|
||||
|
||||
@@ -13,6 +13,7 @@ DEFAULT_CHECKS = [
|
||||
("training_case_package", [sys.executable, "scripts/run_training_case_package.py"]),
|
||||
("migration_static_checks", [sys.executable, "scripts/run_migration_static_checks.py"]),
|
||||
("script_static_checks", [sys.executable, "scripts/run_script_static_checks.py"]),
|
||||
("trading_delivery_readiness", [sys.executable, "scripts/check_trading_delivery_readiness.py"]),
|
||||
("dashboard_js_syntax", ["node", "--check", "src/web/app.js"]),
|
||||
]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user