from __future__ import annotations
|
|
import argparse
|
import csv
|
import json
|
import math
|
from collections import Counter
|
from datetime import datetime
|
from pathlib import Path
|
from typing import Any
|
|
import pandas as pd
|
|
from retry_public_5m_proxy_baostock_per_request import PACKAGE_ROOT, cache_path, date_key, fnum, local_daily_high, time_hhmm
|
|
|
STRICT_ROOT = PACKAGE_ROOT.parent / "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001"
|
LOCAL_MINUTE_ROOT = Path("E:/quant/2023_front_m")
|
LOCAL_DAILY_ROOT = Path("E:/quant/a_share_daily_front_20230101_20260508_complete/daily")
|
|
LOT_LEDGER = STRICT_ROOT / "strict_position_lot_ledger.csv"
|
SELL_SIGNAL_LEDGER = STRICT_ROOT / "strict_sell_signal_ledger.csv"
|
COVERAGE_CSV = PACKAGE_ROOT / "open_volume_stall_scan_coverage.csv"
|
CANDIDATE_CSV = PACKAGE_ROOT / "open_volume_stall_independent_candidates.csv"
|
SUMMARY_JSON = PACKAGE_ROOT / "open_volume_stall_scan_summary.json"
|
SUMMARY_MD = PACKAGE_ROOT / "open_volume_stall_scan_summary.md"
|
PROGRESS_JSON = PACKAGE_ROOT / "open_volume_stall_scan_progress.json"
|
|
|
def safe_float(value: Any) -> float:
|
try:
|
if pd.isna(value) or str(value).strip() == "":
|
return math.nan
|
return float(value)
|
except Exception:
|
return math.nan
|
|
|
def market_code(symbol: str) -> tuple[str, str]:
|
code, market = symbol.split(".")
|
return code, market
|
|
|
def minute_path(symbol: str) -> Path:
|
code, market = market_code(symbol)
|
return LOCAL_MINUTE_ROOT / market / f"price_{code}.csv"
|
|
|
def daily_path(symbol: str) -> Path:
|
return LOCAL_DAILY_ROOT / f"{symbol}.csv"
|
|
|
def normalize_time(value: str) -> str:
|
text = str(value).strip()
|
if not text:
|
return ""
|
if " " in text:
|
text = text.split()[-1]
|
parts = text.split(":")
|
if len(parts) == 2:
|
return f"{int(parts[0]):02d}:{int(parts[1]):02d}:00"
|
if len(parts) >= 3:
|
return f"{int(parts[0]):02d}:{int(parts[1]):02d}:{int(float(parts[2])):02d}"
|
return text
|
|
|
def load_daily(symbol: str, cache: dict[str, pd.DataFrame]) -> pd.DataFrame:
|
if symbol in cache:
|
return cache[symbol]
|
path = daily_path(symbol)
|
if not path.exists():
|
df = pd.DataFrame()
|
else:
|
df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig")
|
df["date_key"] = df["trade_date"].map(date_key)
|
for col in ["open", "high", "low", "close", "preClose", "volume"]:
|
if col in df.columns:
|
df[f"{col}_num"] = pd.to_numeric(df[col], errors="coerce")
|
df = df.sort_values("date_key")
|
cache[symbol] = df
|
return df
|
|
|
def load_minute(symbol: str, cache: dict[str, pd.DataFrame]) -> pd.DataFrame:
|
if symbol in cache:
|
return cache[symbol]
|
path = minute_path(symbol)
|
if not path.exists() or path.stat().st_size == 0:
|
df = pd.DataFrame()
|
else:
|
df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig")
|
df["date_key"] = df["timetag"].str.slice(0, 8)
|
df["trade_time"] = df["timetag"].str.slice(9, 17)
|
df = df.rename(columns={"volumn": "volume"})
|
for col in ["open", "high", "low", "close", "volume", "amount"]:
|
if col in df.columns:
|
df[f"{col}_num"] = pd.to_numeric(df[col], errors="coerce")
|
df = df.sort_values(["date_key", "trade_time"])
|
cache[symbol] = df
|
return df
|
|
|
def trade_dates_for_lot(daily: pd.DataFrame, start_date: str, end_date: str, max_open_days: int) -> list[str]:
|
if daily.empty:
|
return []
|
start = date_key(start_date)
|
end = date_key(end_date) if end_date else str(daily["date_key"].max())
|
rows = daily[(daily["date_key"] >= start) & (daily["date_key"] <= end)]["date_key"].tolist()
|
return rows[:max_open_days] if not end_date else rows
|
|
|
def previous_dates(daily: pd.DataFrame, d: str, n: int) -> list[str]:
|
dates = daily[daily["date_key"] < d]["date_key"].tolist()
|
return dates[-n:]
|
|
|
def daily_prev_close(daily: pd.DataFrame, d: str) -> float:
|
prevs = previous_dates(daily, d, 1)
|
if not prevs:
|
return math.nan
|
row = daily[daily["date_key"] == prevs[-1]]
|
return safe_float(row.iloc[0].get("close_num", math.nan)) if not row.empty else math.nan
|
|
|
def scale_for_day(symbol: str, d: str, day_high: float, daily_cache: dict[str, pd.DataFrame]) -> tuple[float, str]:
|
daily_high = local_daily_high(symbol, f"{d[:4]}-{d[4:6]}-{d[6:8]}", daily_cache)
|
if daily_high is None or not day_high or math.isnan(day_high) or day_high <= 0:
|
return 1.0, "RAW_PRICE_NO_SCALE"
|
return daily_high / day_high, "SCALED_TO_LOCAL_DAILY_HIGH"
|
|
|
def exact_1m_metrics(symbol: str, d: str, daily: pd.DataFrame, minute: pd.DataFrame, daily_cache: dict[str, pd.DataFrame]) -> dict[str, Any]:
|
if minute.empty:
|
return {"scan_status": "NO_LOCAL_1M_FILE"}
|
day = minute[minute["date_key"] == d]
|
if day.empty:
|
return {"scan_status": "NO_LOCAL_1M_DATE"}
|
first = day.iloc[0]
|
prev_close = daily_prev_close(daily, d)
|
if math.isnan(prev_close) or prev_close <= 0:
|
return {"scan_status": "NO_DAILY_PREV_CLOSE"}
|
scale, source = scale_for_day(symbol, d, safe_float(day["high_num"].max()), daily_cache)
|
first_close_raw = safe_float(first.get("close_num"))
|
first_close = first_close_raw * scale if not math.isnan(first_close_raw) else math.nan
|
first_gain = first_close / prev_close - 1 if not math.isnan(first_close) else math.nan
|
vols = []
|
for pd_key in previous_dates(daily, d, 5):
|
prev_day = minute[minute["date_key"] == pd_key]
|
if not prev_day.empty:
|
vols.append(safe_float(prev_day.iloc[0].get("volume_num")))
|
vols = [v for v in vols if not math.isnan(v) and v > 0]
|
avg_first_vol = float(pd.Series(vols).mean()) if vols else math.nan
|
first_vol = safe_float(first.get("volume_num"))
|
vol_ratio = first_vol / avg_first_vol if avg_first_vol and not math.isnan(avg_first_vol) and avg_first_vol > 0 else math.nan
|
hit = (not math.isnan(vol_ratio)) and vol_ratio >= 10 and (not math.isnan(first_gain)) and 0.01 <= first_gain <= 0.02
|
return {
|
"scan_status": "SCANNED_EXACT_1M",
|
"scan_basis": "EXACT_1M",
|
"price_source": source,
|
"first_bar_time": first.get("trade_time", ""),
|
"first_close_raw": first_close_raw,
|
"first_close_scaled": first_close,
|
"prev_close_daily": prev_close,
|
"first_gain_pct": first_gain,
|
"first_volume": first_vol,
|
"avg_previous_first_volume": avg_first_vol,
|
"open_volume_ratio": vol_ratio,
|
"stall_hit": hit,
|
}
|
|
|
def load_5m_day(symbol: str, d: str) -> pd.DataFrame:
|
path = cache_path(symbol, f"{d[:4]}-{d[4:6]}-{d[6:8]}")
|
if not path.exists() or path.stat().st_size == 0:
|
return pd.DataFrame()
|
df = pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig")
|
if df.empty:
|
return df
|
df["hhmm"] = df["time"].map(time_hhmm)
|
for col in ["open", "high", "low", "close", "volume", "amount"]:
|
if col in df.columns:
|
df[f"{col}_num"] = pd.to_numeric(df[col], errors="coerce")
|
return df.sort_values("hhmm")
|
|
|
def proxy_5m_metrics(symbol: str, d: str, daily: pd.DataFrame, daily_cache: dict[str, pd.DataFrame]) -> dict[str, Any]:
|
day = load_5m_day(symbol, d)
|
if day.empty:
|
return {"scan_status": "NO_5M_PROXY_CACHE"}
|
prev_close = daily_prev_close(daily, d)
|
if math.isnan(prev_close) or prev_close <= 0:
|
return {"scan_status": "NO_DAILY_PREV_CLOSE"}
|
first = day.iloc[0]
|
scale, source = scale_for_day(symbol, d, safe_float(day["high_num"].max()), daily_cache)
|
first_close_raw = safe_float(first.get("close_num"))
|
first_close = first_close_raw * scale if not math.isnan(first_close_raw) else math.nan
|
first_gain = first_close / prev_close - 1 if not math.isnan(first_close) else math.nan
|
vols = []
|
missing_prev = 0
|
for pd_key in previous_dates(daily, d, 5):
|
prev_day = load_5m_day(symbol, pd_key)
|
if prev_day.empty:
|
missing_prev += 1
|
else:
|
vols.append(safe_float(prev_day.iloc[0].get("volume_num")))
|
vols = [v for v in vols if not math.isnan(v) and v > 0]
|
if not vols:
|
return {"scan_status": "NO_PROXY_PREVIOUS_5M_VOLUME", "scan_basis": "PROXY_5M", "missing_previous_5m_days": missing_prev}
|
avg_first_vol = float(pd.Series(vols).mean())
|
first_vol = safe_float(first.get("volume_num"))
|
vol_ratio = first_vol / avg_first_vol if avg_first_vol > 0 else math.nan
|
hit = (not math.isnan(vol_ratio)) and vol_ratio >= 10 and (not math.isnan(first_gain)) and 0.01 <= first_gain <= 0.02
|
return {
|
"scan_status": "SCANNED_PROXY_5M",
|
"scan_basis": "PROXY_5M",
|
"price_source": source,
|
"first_bar_time": first.get("hhmm", ""),
|
"first_close_raw": first_close_raw,
|
"first_close_scaled": first_close,
|
"prev_close_daily": prev_close,
|
"first_gain_pct": first_gain,
|
"first_volume": first_vol,
|
"avg_previous_first_volume": avg_first_vol,
|
"open_volume_ratio": vol_ratio,
|
"stall_hit": hit,
|
"missing_previous_5m_days": missing_prev,
|
}
|
|
|
def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
|
keys: list[str] = []
|
seen: set[str] = set()
|
for row in rows:
|
for key in row:
|
if key not in seen:
|
seen.add(key)
|
keys.append(key)
|
with path.open("w", newline="", encoding="utf-8-sig") as f:
|
writer = csv.DictWriter(f, fieldnames=keys, extrasaction="ignore")
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def write_progress(payload: dict[str, Any]) -> None:
|
PROGRESS_JSON.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
def main() -> None:
|
parser = argparse.ArgumentParser()
|
parser.add_argument("--max-open-lot-days", type=int, default=10)
|
parser.add_argument("--use-proxy-5m", action="store_true")
|
args = parser.parse_args()
|
|
lots = pd.read_csv(LOT_LEDGER, dtype=str, keep_default_na=False, encoding="utf-8-sig")
|
signals = pd.read_csv(SELL_SIGNAL_LEDGER, dtype=str, keep_default_na=False, encoding="utf-8-sig")
|
signal_by_id = signals.set_index("signal_id").to_dict("index") if "signal_id" in signals.columns else {}
|
daily_cache: dict[str, pd.DataFrame] = {}
|
coverage: list[dict[str, Any]] = []
|
candidates: list[dict[str, Any]] = []
|
|
grouped = list(lots.groupby("symbol", sort=True))
|
for symbol_index, (symbol, symbol_lots) in enumerate(grouped, start=1):
|
daily = load_daily(symbol, daily_cache)
|
minute = load_minute(symbol, {})
|
for lot in symbol_lots.to_dict("records"):
|
dates = trade_dates_for_lot(
|
daily,
|
lot["sellable_from_trade_date"],
|
lot.get("exit_trade_date", ""),
|
args.max_open_lot_days,
|
)
|
exit_time = normalize_time(lot.get("exit_time", ""))
|
exit_signal = signal_by_id.get(lot.get("exit_signal_id", ""), {})
|
for d in dates:
|
metrics = exact_1m_metrics(symbol, d, daily, minute, daily_cache)
|
if metrics.get("scan_status", "").startswith("NO_") and args.use_proxy_5m:
|
metrics = proxy_5m_metrics(symbol, d, daily, daily_cache)
|
first_time = normalize_time(str(metrics.get("first_bar_time", "")))
|
if (
|
lot.get("exit_trade_date")
|
and date_key(lot["exit_trade_date"]) == d
|
and exit_time
|
and first_time
|
and exit_time <= first_time
|
):
|
metrics = {"scan_status": "SKIP_EXIT_BEFORE_OPEN_BAR", **metrics}
|
|
base = {
|
"strict_lot_id": lot.get("strict_lot_id", ""),
|
"source_lot_id": lot.get("source_lot_id", ""),
|
"case_id": lot.get("case_id", ""),
|
"symbol": symbol,
|
"scan_trade_date": f"{d[:4]}-{d[4:6]}-{d[6:8]}",
|
"entry_trade_date": lot.get("entry_trade_date", ""),
|
"sellable_from_trade_date": lot.get("sellable_from_trade_date", ""),
|
"exit_trade_date": lot.get("exit_trade_date", ""),
|
"exit_time": lot.get("exit_time", ""),
|
"existing_exit_signal_id": lot.get("exit_signal_id", ""),
|
"existing_exit_signal_type": exit_signal.get("signal_type", ""),
|
}
|
row = {**base, **metrics}
|
coverage.append(row)
|
if str(row.get("stall_hit")) == "True":
|
candidates.append(row)
|
if symbol_index % 50 == 0:
|
write_progress(
|
{
|
"status": "RUNNING",
|
"symbol_index": symbol_index,
|
"symbol_count": len(grouped),
|
"coverage_rows": len(coverage),
|
"candidate_rows": len(candidates),
|
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
}
|
)
|
|
write_csv(COVERAGE_CSV, coverage)
|
write_csv(CANDIDATE_CSV, candidates)
|
summary = {
|
"generated_at": datetime.now().isoformat(timespec="seconds"),
|
"lot_rows": len(lots),
|
"coverage_rows": len(coverage),
|
"candidate_rows": len(candidates),
|
"coverage_status_counts": dict(Counter(str(row.get("scan_status", "")) for row in coverage)),
|
"candidate_basis_counts": dict(Counter(str(row.get("scan_basis", "")) for row in candidates)),
|
"threshold": "open_volume_ratio >= 10 and 0.01 <= first_gain_pct <= 0.02",
|
"boundaries": [
|
"EXACT_1M rows use local 1-minute files and daily previous close, with prices scaled to local daily high when possible.",
|
"PROXY_5M rows use cached public 5-minute bars and are not strict opening-one-minute evidence.",
|
"This is an independent scan; candidate rows indicate possible omissions that require chart/manual review before changing V1 ledgers.",
|
],
|
}
|
SUMMARY_JSON.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
SUMMARY_MD.write_text(
|
"\n".join(
|
[
|
"# Open Volume Stall Independent Scan",
|
"",
|
f"- generated_at: {summary['generated_at']}",
|
f"- lot_rows: {summary['lot_rows']}",
|
f"- coverage_rows: {summary['coverage_rows']}",
|
f"- candidate_rows: {summary['candidate_rows']}",
|
f"- coverage_status_counts: {json.dumps(summary['coverage_status_counts'], ensure_ascii=False)}",
|
f"- candidate_basis_counts: {json.dumps(summary['candidate_basis_counts'], ensure_ascii=False)}",
|
f"- threshold: {summary['threshold']}",
|
"",
|
"## Boundaries",
|
*[f"- {item}" for item in summary["boundaries"]],
|
"",
|
]
|
),
|
encoding="utf-8",
|
)
|
write_progress({"status": "FINISHED", **summary})
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
