"""Scan local A-share K-line/profile data for market reverse-gap candidates.

The script uses local MySQL read-only tables:

- a_share_profile_snapshot
- a_share_daily_price

It writes market manifestation audit artifacts into the industry case layout.
Use the result as a gap-finding input, not as an investment conclusion.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import os
import re
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path


SCRIPT_PATH = Path(__file__).resolve()
PROJECT_ROOT = SCRIPT_PATH.parents[2]
DEFAULT_MYSQL_EXE = Path(os.environ.get("TIANXIA_MYSQL_EXE", "M:/mysql/server/bin/mysql.exe"))


def now_iso() -> str:
    return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


def safe_name(value: str, fallback: str = "artifact", max_len: int = 96) -> str:
    value = re.sub(r"[^A-Za-z0-9._-]+", "_", value.strip())
    value = value.strip("._-")
    return (value or fallback)[:max_len]


def sql_literal(value: object) -> str:
    if value is None:
        return "NULL"
    text = str(value)
    return "'" + text.replace("\\", "\\\\").replace("'", "''") + "'"


def sha256_text(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def project_relative(path: Path) -> str:
    try:
        return path.resolve().relative_to(PROJECT_ROOT.resolve()).as_posix()
    except ValueError:
        return path.resolve().as_posix()


def keyword_predicate(keywords: list[str]) -> str:
    clauses = []
    for keyword in keywords:
        pattern = "%" + keyword.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%"
        lit = sql_literal(pattern)
        clauses.append(
            "("
            f"p.industry_l1 LIKE {lit} ESCAPE '\\\\' OR "
            f"p.industry_l2 LIKE {lit} ESCAPE '\\\\' OR "
            f"p.concepts_raw LIKE {lit} ESCAPE '\\\\' OR "
            f"p.main_business_raw LIKE {lit} ESCAPE '\\\\' OR "
            f"p.stock_name LIKE {lit} ESCAPE '\\\\'"
            ")"
        )
    return " OR ".join(clauses)


def build_sql(args: argparse.Namespace) -> str:
    if not args.keyword:
        raise SystemExit("pass at least one --keyword")
    pred = keyword_predicate(args.keyword)
    start = sql_literal(args.start_date)
    end = sql_literal(args.end_date)
    min_return = float(args.min_return_pct)
    strong_day_pct = float(args.strong_day_pct)
    amount_ratio = float(args.amount_ratio)
    return f"""
WITH latest_profile AS (
  SELECT MAX(snapshot_date) AS snapshot_date
  FROM a_share_profile_snapshot
  WHERE snapshot_date <= {end}
),
universe AS (
  SELECT
    p.symbol,
    p.stock_name,
    p.industry_l1,
    p.industry_l2,
    p.concepts_raw,
    p.main_business_raw
  FROM a_share_profile_snapshot p
  JOIN latest_profile lp ON lp.snapshot_date = p.snapshot_date
  WHERE p.status = 'OK'
    AND ({pred})
),
price_with_prev AS (
  SELECT
    d.symbol,
    d.trade_date,
    d.open_price,
    d.high_price,
    d.low_price,
    d.close_price,
    d.volume,
    d.amount,
    LAG(d.close_price) OVER (PARTITION BY d.symbol ORDER BY d.trade_date) AS prev_close,
    AVG(d.amount) OVER (
      PARTITION BY d.symbol
      ORDER BY d.trade_date
      ROWS BETWEEN 20 PRECEDING AND 1 PRECEDING
    ) AS avg_amount_20
  FROM a_share_daily_price d
  JOIN universe u ON u.symbol = d.symbol
  WHERE d.trade_date BETWEEN DATE_SUB({start}, INTERVAL 30 DAY) AND {end}
),
window_rows AS (
  SELECT
    u.stock_name,
    u.industry_l1,
    u.industry_l2,
    u.concepts_raw,
    u.main_business_raw,
    p.*,
    CASE
      WHEN p.prev_close IS NULL OR p.prev_close = 0 THEN NULL
      ELSE (p.close_price / p.prev_close - 1) * 100
    END AS daily_return_pct,
    CASE
      WHEN p.avg_amount_20 IS NULL OR p.avg_amount_20 = 0 THEN NULL
      ELSE p.amount / p.avg_amount_20
    END AS amount_ratio_20
  FROM price_with_prev p
  JOIN universe u ON u.symbol = p.symbol
  WHERE p.trade_date BETWEEN {start} AND {end}
),
ranked AS (
  SELECT
    w.*,
    ROW_NUMBER() OVER (PARTITION BY w.symbol ORDER BY w.trade_date) AS rn_first,
    ROW_NUMBER() OVER (PARTITION BY w.symbol ORDER BY w.trade_date DESC) AS rn_last
  FROM window_rows w
),
agg AS (
  SELECT
    symbol,
    MAX(stock_name) AS company_name,
    MAX(industry_l1) AS industry_l1,
    MAX(industry_l2) AS industry_l2,
    MAX(concepts_raw) AS concepts_raw,
    MAX(main_business_raw) AS main_business_raw,
    MIN(trade_date) AS window_start,
    MAX(trade_date) AS window_end,
    MAX(CASE WHEN rn_first = 1 THEN close_price END) AS first_close,
    MAX(CASE WHEN rn_last = 1 THEN close_price END) AS last_close,
    COUNT(*) AS trading_days,
    SUM(CASE WHEN daily_return_pct >= {strong_day_pct} THEN 1 ELSE 0 END) AS strong_up_days,
    MAX(daily_return_pct) AS max_daily_return_pct,
    MAX(amount_ratio_20) AS max_amount_ratio_20,
    MAX(amount) AS max_amount
  FROM ranked
  GROUP BY symbol
),
scored AS (
  SELECT
    *,
    CASE
      WHEN first_close IS NULL OR first_close = 0 THEN NULL
      ELSE (last_close / first_close - 1) * 100
    END AS period_return_pct
  FROM agg
)
SELECT
  {sql_literal(args.scan_id)} AS scan_id,
  {sql_literal(args.industry_case)} AS industry_case,
  symbol,
  company_name,
  industry_l1,
  industry_l2,
  window_start,
  window_end,
  trading_days,
  ROUND(first_close, 4) AS first_close,
  ROUND(last_close, 4) AS last_close,
  ROUND(period_return_pct, 4) AS period_return_pct,
  strong_up_days,
  ROUND(max_daily_return_pct, 4) AS max_daily_return_pct,
  ROUND(max_amount_ratio_20, 4) AS max_amount_ratio_20,
  CASE
    WHEN period_return_pct >= {min_return} OR strong_up_days >= 1 OR max_amount_ratio_20 >= {amount_ratio}
    THEN 'STRONG_MANIFESTATION'
    ELSE 'WEAK_OR_NORMAL'
  END AS manifestation_type,
  'REVIEW_REQUIRED' AS scope_type,
  'REVIEW_REQUIRED' AS suspected_missing_track,
  'MARKET_REVERSE_GAP_SCAN' AS gap_type,
  CASE
    WHEN period_return_pct >= {min_return} OR strong_up_days >= 1 OR max_amount_ratio_20 >= {amount_ratio}
    THEN 1
    ELSE 0
  END AS supplement_required_flag,
  'Review core business against industry scheme before adding to evidence queue' AS supplement_question,
  'DRAFT_FOR_REVIEW' AS review_status,
  LEFT(concepts_raw, 500) AS concepts_excerpt,
  LEFT(main_business_raw, 500) AS main_business_excerpt
FROM scored
WHERE period_return_pct >= {min_return}
   OR strong_up_days >= 1
   OR max_amount_ratio_20 >= {amount_ratio}
ORDER BY supplement_required_flag DESC, period_return_pct DESC, strong_up_days DESC, max_amount_ratio_20 DESC;
""".strip()


def run_mysql(args: argparse.Namespace, sql: str) -> str:
    password = os.environ.get("TIANXIA_MYSQL_PASSWORD")
    if not password:
        raise SystemExit("TIANXIA_MYSQL_PASSWORD is required in the environment")
    mysql_exe = Path(args.mysql_exe)
    if not mysql_exe.exists():
        raise SystemExit(f"mysql executable not found: {mysql_exe}")
    cmd = [
        str(mysql_exe),
        "--protocol=TCP",
        "--batch",
        "--raw",
        "--quick",
        "--default-character-set=utf8mb4",
        f"--host={args.host}",
        f"--port={args.port}",
        f"--user={args.user}",
        args.database,
        "-e",
        sql,
    ]
    env = os.environ.copy()
    env["MYSQL_PWD"] = password
    proc = subprocess.run(cmd, text=True, capture_output=True, env=env, check=False)
    if proc.returncode != 0:
        raise SystemExit(proc.stderr.strip() or proc.stdout.strip())
    return proc.stdout


def write_csv_from_tsv(tsv: str, out_path: Path) -> int:
    out_path.parent.mkdir(parents=True, exist_ok=True)
    row_count = 0
    with out_path.open("w", encoding="utf-8-sig", newline="") as fh:
        writer = csv.writer(fh)
        for line_index, line in enumerate(tsv.splitlines()):
            writer.writerow(line.split("\t"))
            if line_index > 0:
                row_count += 1
    return row_count


def append_manifest(path: Path, row: dict[str, str]) -> None:
    exists = path.exists()
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8-sig", newline="") as fh:
        writer = csv.DictWriter(fh, fieldnames=list(row.keys()))
        if not exists:
            writer.writeheader()
        writer.writerow(row)


def scan(args: argparse.Namespace) -> int:
    if not args.scan_id:
        args.scan_id = f"{safe_name(args.run_id, 'run')}_MARKET_GAP_SCAN"
    industry_root = PROJECT_ROOT / "ana-data" / "cases" / args.industry_case
    supplement_dir = industry_root / "supplement"
    manifest_dir = industry_root / "manifest"
    supplement_dir.mkdir(parents=True, exist_ok=True)
    manifest_dir.mkdir(parents=True, exist_ok=True)

    run_safe = safe_name(args.run_id, "run")
    sql = build_sql(args)
    sql_hash = sha256_text(sql)
    audit_path = supplement_dir / f"market_manifestation_gap_audit_{run_safe}_{sql_hash[:12]}.csv"
    summary_path = supplement_dir / f"market_manifestation_gap_summary_{run_safe}_{sql_hash[:12]}.json"
    sql_path = manifest_dir / f"market_manifestation_gap_query_{run_safe}_{sql_hash[:12]}.sql"
    manifest_path = manifest_dir / f"market_manifestation_gap_manifest_{run_safe}.csv"

    tsv = run_mysql(args, sql)
    row_count = write_csv_from_tsv(tsv, audit_path)
    sql_path.write_text(sql + "\n", encoding="utf-8")
    summary = {
        "scan_id": args.scan_id,
        "case_id": args.case_id,
        "batch_id": args.batch_id,
        "run_id": args.run_id,
        "industry_case": args.industry_case,
        "start_date": args.start_date,
        "end_date": args.end_date,
        "keywords": args.keyword,
        "min_return_pct": args.min_return_pct,
        "strong_day_pct": args.strong_day_pct,
        "amount_ratio": args.amount_ratio,
        "row_count": row_count,
        "audit_path": project_relative(audit_path),
        "sql_sha256": sql_hash,
        "generated_at": now_iso(),
        "readout": "GAP_SCAN_ONLY_NOT_INVESTMENT_CONCLUSION",
    }
    summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
    append_manifest(
        manifest_path,
        {
            "scan_id": args.scan_id,
            "case_id": args.case_id,
            "batch_id": args.batch_id,
            "run_id": args.run_id,
            "industry_case": args.industry_case,
            "start_date": args.start_date,
            "end_date": args.end_date,
            "keywords": "|".join(args.keyword),
            "sql_sha256": sql_hash,
            "sql_relative_path": project_relative(sql_path),
            "audit_relative_path": project_relative(audit_path),
            "summary_relative_path": project_relative(summary_path),
            "row_count": str(row_count),
            "generated_at": summary["generated_at"],
            "review_status": "DRAFT_FOR_REVIEW",
        },
    )
    print(f"audit={project_relative(audit_path)} rows={row_count}")
    print(f"summary={project_relative(summary_path)}")
    return 0


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--industry-case", required=True)
    parser.add_argument("--case-id", required=True)
    parser.add_argument("--batch-id", required=True)
    parser.add_argument("--run-id", required=True)
    parser.add_argument("--scan-id")
    parser.add_argument("--start-date", required=True)
    parser.add_argument("--end-date", required=True)
    parser.add_argument("--keyword", action="append", required=True)
    parser.add_argument("--min-return-pct", type=float, default=20.0)
    parser.add_argument("--strong-day-pct", type=float, default=9.5)
    parser.add_argument("--amount-ratio", type=float, default=2.0)
    parser.add_argument("--mysql-exe", default=str(DEFAULT_MYSQL_EXE))
    parser.add_argument("--host", default=os.environ.get("TIANXIA_MYSQL_HOST", "127.0.0.1"))
    parser.add_argument("--port", default=os.environ.get("TIANXIA_MYSQL_PORT", "3306"))
    parser.add_argument("--user", default=os.environ.get("TIANXIA_MYSQL_USER", "root"))
    parser.add_argument("--database", default=os.environ.get("TIANXIA_MYSQL_DB", "tianxia"))
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv or sys.argv[1:])
    return scan(args)


if __name__ == "__main__":
    raise SystemExit(main())