from __future__ import annotations
|
|
import csv
|
import hashlib
|
import importlib.util
|
import json
|
from datetime import datetime
|
from pathlib import Path
|
from zoneinfo import ZoneInfo
|
|
import pandas as pd
|
|
|
RUN_ID = "RUN-ANA-WUJI-RECENT-STRICT-CANDIDATES-20260611-001"
|
ROOT = Path(__file__).resolve().parents[1]
|
PROJECT_ROOT = ROOT.parents[2]
|
CHART_DIR = ROOT / "charts"
|
SOURCE_SCRIPT = (
|
PROJECT_ROOT
|
/ "ana-data/result/RUN-ANA-WUJI-FULL-2023-2026-20260608-001/tools/generate_candidate_pool.py"
|
)
|
|
|
def load_source_module():
|
spec = importlib.util.spec_from_file_location("wuji_candidate_source", SOURCE_SCRIPT)
|
if spec is None or spec.loader is None:
|
raise RuntimeError(f"Unable to load source script: {SOURCE_SCRIPT}")
|
module = importlib.util.module_from_spec(spec)
|
spec.loader.exec_module(module)
|
return module
|
|
|
def sha256_file(path: Path) -> str:
|
h = hashlib.sha256()
|
with path.open("rb") as f:
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
h.update(chunk)
|
return h.hexdigest()
|
|
|
def write_csv(df: pd.DataFrame, name: str) -> Path:
|
path = ROOT / name
|
df.to_csv(path, index=False, encoding="utf-8-sig")
|
return path
|
|
|
def add_features(daily: pd.DataFrame) -> pd.DataFrame:
|
daily = daily.sort_values(["symbol", "trade_date"]).reset_index(drop=True)
|
grouped = daily.groupby("symbol", group_keys=False)
|
daily["prev_close"] = grouped["close_price"].shift(1)
|
daily["prev5_avg_volume"] = grouped["volume"].transform(
|
lambda s: s.shift(1).rolling(5, min_periods=5).mean()
|
)
|
daily["prev10_avg_volume"] = grouped["volume"].transform(
|
lambda s: s.shift(1).rolling(10, min_periods=10).mean()
|
)
|
daily["volume_ratio"] = daily["volume"] / daily["prev5_avg_volume"]
|
daily["volume_ratio_ma10"] = daily["volume"] / daily["prev10_avg_volume"]
|
daily["daily_return_pct"] = (daily["close_price"] / daily["prev_close"] - 1.0) * 100.0
|
daily["limit_up_event_flag"] = daily["prev_close"].gt(0) & (
|
daily["high_price"] / daily["prev_close"] - 1.0
|
).ge(0.095)
|
daily["upper_shadow_pct"] = (
|
(daily["high_price"] - daily[["open_price", "close_price"]].max(axis=1))
|
/ daily["prev_close"]
|
) * 100.0
|
day_range = daily["high_price"] - daily["low_price"]
|
daily["upper_shadow_range_ratio"] = (
|
daily["high_price"] - daily[["open_price", "close_price"]].max(axis=1)
|
) / day_range.replace(0, pd.NA)
|
daily["body_pct"] = (
|
(daily["close_price"] - daily["open_price"]).abs() / daily["prev_close"]
|
) * 100.0
|
daily["prev60_high"] = grouped["high_price"].transform(
|
lambda s: s.shift(1).rolling(60, min_periods=20).max()
|
)
|
daily["prev60_low"] = grouped["low_price"].transform(
|
lambda s: s.shift(1).rolling(60, min_periods=20).min()
|
)
|
daily["flat60_range_pct"] = (daily["prev60_high"] / daily["prev60_low"] - 1.0) * 100.0
|
|
prior_limitup_30 = []
|
last_limitup_date = []
|
days_since_limitup = []
|
last_limitup_close = []
|
post_limitup_min_low = []
|
prev_high_volume = []
|
prev_high_ref_date = []
|
prev_high_ref_policy = []
|
|
for _symbol, group in daily.groupby("symbol", sort=False):
|
highs = group["high_price"].to_numpy()
|
lows = group["low_price"].to_numpy()
|
closes = group["close_price"].to_numpy()
|
vols = group["volume"].to_numpy()
|
dates = group["trade_date"].to_numpy()
|
limit_flags = group["limit_up_event_flag"].to_numpy()
|
n = len(group)
|
for i in range(n):
|
start30 = max(0, i - 30)
|
prior_limit_idx = [j for j in range(start30, i) if limit_flags[j]]
|
if prior_limit_idx:
|
last = prior_limit_idx[-1]
|
prior_limitup_30.append(True)
|
last_limitup_date.append(pd.Timestamp(dates[last]))
|
days_since_limitup.append(i - last)
|
last_limitup_close.append(closes[last])
|
post_limitup_min_low.append(float(lows[last + 1 : i + 1].min()))
|
else:
|
prior_limitup_30.append(False)
|
last_limitup_date.append(pd.NaT)
|
days_since_limitup.append(pd.NA)
|
last_limitup_close.append(float("nan"))
|
post_limitup_min_low.append(float("nan"))
|
|
start60 = max(0, i - 60)
|
if i - start60 >= 20:
|
window_highs = highs[start60:i]
|
max_pos = int(window_highs.argmax())
|
prev_high_volume.append(vols[start60 + max_pos])
|
prev_high_ref_date.append(pd.Timestamp(dates[start60 + max_pos]))
|
prev_high_ref_policy.append("FIRST_PREVIOUS_HIGH_IN_60D_WINDOW")
|
else:
|
prev_high_volume.append(float("nan"))
|
prev_high_ref_date.append(pd.NaT)
|
prev_high_ref_policy.append("")
|
|
daily["prior_limitup_30_flag"] = prior_limitup_30
|
daily["last_prior_limitup_date"] = last_limitup_date
|
daily["days_since_prior_limitup"] = days_since_limitup
|
daily["last_prior_limitup_close"] = last_limitup_close
|
daily["post_limitup_min_low"] = post_limitup_min_low
|
daily["pullback_from_last_limitup_close_pct"] = (
|
daily["post_limitup_min_low"] / daily["last_prior_limitup_close"] - 1.0
|
) * 100.0
|
daily["prev60_high_volume"] = prev_high_volume
|
daily["prev60_high_ref_date"] = prev_high_ref_date
|
daily["prev60_high_ref_policy"] = prev_high_ref_policy
|
daily["touch_prev_high_flag"] = daily["prev60_high"].gt(0) & daily["high_price"].ge(
|
daily["prev60_high"] * 0.995
|
)
|
daily["prev_high_volume_pass_flag"] = (~daily["touch_prev_high_flag"]) | daily[
|
"volume"
|
].gt(daily["prev60_high_volume"])
|
return daily
|
|
|
def breadth_from_daily(daily: pd.DataFrame) -> pd.DataFrame:
|
gate = daily[daily["prev_close"].gt(0)].copy()
|
gate["up_flag"] = gate["close_price"].gt(gate["prev_close"])
|
gate["flat_flag"] = gate["close_price"].eq(gate["prev_close"])
|
gate["down_flag"] = gate["close_price"].lt(gate["prev_close"])
|
breadth = (
|
gate.groupby("trade_date")
|
.agg(
|
stock_count=("symbol", "count"),
|
up_count=("up_flag", "sum"),
|
flat_count=("flat_flag", "sum"),
|
down_count=("down_flag", "sum"),
|
)
|
.reset_index()
|
.rename(columns={"trade_date": "signal_trade_date"})
|
)
|
breadth["market_gate_open_flag"] = breadth["up_count"].ge(3000)
|
breadth["market_gate_status"] = breadth["market_gate_open_flag"].map(
|
{
|
True: "MKT_GATE_OPEN_SIGNAL_DAY_UP_3000_RECALC_FROM_DAILY",
|
False: "NO_TRADE_MARKET_GATE_CLOSED_SIGNAL_DAY_RECALC_FROM_DAILY",
|
}
|
)
|
return breadth
|
|
|
def build_strict_candidates(
|
daily: pd.DataFrame,
|
signal_date: pd.Timestamp,
|
entry_date: pd.Timestamp | None,
|
label: str,
|
breadth: pd.DataFrame,
|
) -> pd.DataFrame:
|
frame = daily[daily["trade_date"].eq(signal_date)].copy()
|
frame = frame[~frame["symbol"].str.endswith(".BJ")].copy()
|
frame = frame[
|
frame["prev_close"].gt(0)
|
& frame["prev5_avg_volume"].gt(0)
|
& frame["prior_limitup_30_flag"]
|
& frame["volume_ratio"].ge(2.0)
|
& frame["upper_shadow_pct"].ge(3.0)
|
& frame["upper_shadow_range_ratio"].ge(0.40)
|
& frame["pullback_from_last_limitup_close_pct"].le(-3.0)
|
& frame["prev_high_volume_pass_flag"]
|
].copy()
|
frame["signal_trade_date"] = signal_date
|
frame["entry_trade_date"] = entry_date if entry_date is not None else pd.NaT
|
frame = frame.merge(breadth, on="signal_trade_date", how="left")
|
frame["scan_label"] = label
|
frame["strict_candidate_status"] = "STRICT_CODE_PASS_SUPPORT_REVIEW_REQUIRED"
|
frame["support_manual_decision"] = "SUPPORT_REVIEW_REQUIRED"
|
frame["support_manual_reason_cn"] = "代码硬筛已满足：近30交易日prior涨停、涨停后回调至少3%、严格倍量、长上影、前高量能过滤；底部承接强弱需看图人工确认。"
|
frame["candidate_rank"] = (
|
frame.sort_values(
|
["signal_trade_date", "upper_shadow_pct", "volume_ratio", "amount"],
|
ascending=[True, False, False, False],
|
)
|
.groupby("signal_trade_date")
|
.cumcount()
|
+ 1
|
)
|
frame = frame.sort_values(["candidate_rank", "symbol"]).reset_index(drop=True)
|
frame["candidate_id"] = [f"CAND-{RUN_ID}-{label}-{i + 1:04d}" for i in range(len(frame))]
|
return frame
|
|
|
def draw_chart(history: pd.DataFrame, row: pd.Series, output: Path) -> None:
|
hist = history.tail(110).reset_index(drop=True)
|
width = 1200
|
height = 680
|
left = 70
|
right = 30
|
top = 58
|
price_h = 430
|
vol_top = top + price_h + 35
|
vol_h = 120
|
n = max(len(hist), 1)
|
slot = (width - left - right) / n
|
candle_w = max(2.0, slot * 0.55)
|
low = float(hist["low_price"].min())
|
high = float(hist["high_price"].max())
|
if high <= low:
|
high = low + 1
|
pad = (high - low) * 0.05
|
low -= pad
|
high += pad
|
max_vol = float(hist["volume"].max() or 1)
|
|
def x(i: int) -> float:
|
return left + slot * (i + 0.5)
|
|
def y_price(v: float) -> float:
|
return top + (high - float(v)) / (high - low) * price_h
|
|
def y_vol(v: float) -> float:
|
return vol_top + vol_h - float(v) / max_vol * vol_h
|
|
parts = [
|
f'<svg xmlns="http://www.w3.org/2000/svg" width="{width}" height="{height}" viewBox="0 0 {width} {height}">',
|
'<rect x="0" y="0" width="100%" height="100%" fill="white"/>',
|
f'<text x="{left}" y="28" font-size="18" font-family="Arial" fill="#111">{row["symbol"]} strict candidate {pd.Timestamp(row["signal_trade_date"]).date()} vol={row["volume_ratio"]:.2f} upper={row["upper_shadow_pct"]:.2f}% pullback={row["pullback_from_last_limitup_close_pct"]:.2f}%</text>',
|
f'<rect x="{left}" y="{top}" width="{width-left-right}" height="{price_h}" fill="none" stroke="#cccccc"/>',
|
f'<rect x="{left}" y="{vol_top}" width="{width-left-right}" height="{vol_h}" fill="none" stroke="#dddddd"/>',
|
]
|
for frac in [0, 0.25, 0.5, 0.75, 1.0]:
|
yy = top + frac * price_h
|
price = high - frac * (high - low)
|
parts.append(f'<line x1="{left}" y1="{yy:.1f}" x2="{width-right}" y2="{yy:.1f}" stroke="#eeeeee"/>')
|
parts.append(f'<text x="8" y="{yy+4:.1f}" font-size="11" font-family="Arial" fill="#555">{price:.2f}</text>')
|
|
def ma_path(window: int, color: str) -> None:
|
ma = hist["close_price"].rolling(window).mean()
|
pts = []
|
for i, v in enumerate(ma):
|
if pd.notna(v):
|
pts.append(f"{x(i):.1f},{y_price(float(v)):.1f}")
|
if len(pts) >= 2:
|
parts.append(f'<polyline points="{" ".join(pts)}" fill="none" stroke="{color}" stroke-width="1.3"/>')
|
|
ma_path(5, "#f0a000")
|
ma_path(20, "#1f77b4")
|
ma_path(60, "#9467bd")
|
|
for i, r in hist.iterrows():
|
color = "#d62728" if r["close_price"] >= r["open_price"] else "#2ca02c"
|
cx = x(i)
|
parts.append(
|
f'<line x1="{cx:.1f}" y1="{y_price(r["low_price"]):.1f}" x2="{cx:.1f}" y2="{y_price(r["high_price"]):.1f}" stroke="{color}" stroke-width="1"/>'
|
)
|
y_open = y_price(r["open_price"])
|
y_close = y_price(r["close_price"])
|
rect_y = min(y_open, y_close)
|
rect_h = max(1.0, abs(y_open - y_close))
|
parts.append(
|
f'<rect x="{cx-candle_w/2:.1f}" y="{rect_y:.1f}" width="{candle_w:.1f}" height="{rect_h:.1f}" fill="{color}" opacity="0.9"/>'
|
)
|
vy = y_vol(r["volume"])
|
parts.append(
|
f'<rect x="{cx-candle_w/2:.1f}" y="{vy:.1f}" width="{candle_w:.1f}" height="{vol_top+vol_h-vy:.1f}" fill="{color}" opacity="0.45"/>'
|
)
|
|
signal_idx = hist.index[hist["trade_date"].eq(row["signal_trade_date"])]
|
if len(signal_idx):
|
idx = int(signal_idx[0])
|
cx = x(idx)
|
parts.append(f'<line x1="{cx:.1f}" y1="{top}" x2="{cx:.1f}" y2="{vol_top+vol_h}" stroke="#111" stroke-dasharray="5,4" stroke-width="1.5"/>')
|
parts.append(f'<circle cx="{cx:.1f}" cy="{y_price(row["high_price"]):.1f}" r="5" fill="#ff0000"/>')
|
parts.append(f'<text x="{cx+6:.1f}" y="{top+18}" font-size="12" font-family="Arial" fill="#111">signal</text>')
|
limit_date = row.get("last_prior_limitup_date")
|
if pd.notna(limit_date):
|
limit_idx = hist.index[hist["trade_date"].eq(pd.Timestamp(limit_date))]
|
if len(limit_idx):
|
cx = x(int(limit_idx[0]))
|
parts.append(f'<line x1="{cx:.1f}" y1="{top}" x2="{cx:.1f}" y2="{vol_top+vol_h}" stroke="#ff7f0e" stroke-dasharray="2,3" stroke-width="1.5"/>')
|
parts.append(f'<text x="{cx+6:.1f}" y="{top+36}" font-size="12" font-family="Arial" fill="#ff7f0e">prior limit-up</text>')
|
|
tick_step = max(1, len(hist) // 8)
|
for i in range(0, len(hist), tick_step):
|
parts.append(f'<text x="{x(i)-15:.1f}" y="{height-18}" font-size="11" font-family="Arial" fill="#555" transform="rotate(30 {x(i):.1f},{height-18})">{pd.Timestamp(hist.loc[i, "trade_date"]).strftime("%m-%d")}</text>')
|
parts.append('<text x="960" y="48" font-size="12" font-family="Arial" fill="#f0a000">MA5</text>')
|
parts.append('<text x="1000" y="48" font-size="12" font-family="Arial" fill="#1f77b4">MA20</text>')
|
parts.append('<text x="1050" y="48" font-size="12" font-family="Arial" fill="#9467bd">MA60</text>')
|
parts.append("</svg>")
|
output.write_text("\n".join(parts), encoding="utf-8")
|
|
|
def render_summary(summary: dict) -> str:
|
return f"""# Wuji Recent Strict Candidate Scan
|
|
run_id: `{summary["run_id"]}`
|
generated_at: `{summary["generated_at"]}`
|
|
This package is stricter than the broad non-BJ daily candidate list. It excludes all `.BJ`
|
symbols and requires prior limit-up memory, pullback after that limit-up, strict double-volume,
|
long upper shadow, and previous-high volume guard.
|
|
## Frozen Hard Filters
|
|
- Exclude `.BJ`.
|
- Prior limit-up within 30 trading days, excluding the signal day itself.
|
- Pullback after the prior limit-up: min low after prior limit-up through signal day <= prior limit-up close - 3%.
|
- Strict volume: `volume / previous 5 trading-day average volume >= 2.0`.
|
- Long upper shadow: `upper_shadow_pct >= 3.0` and upper shadow accounts for at least 40% of daily range.
|
- If touching previous 60-trading-day high, signal-day volume must be greater than the previous-high reference volume.
|
|
## Counts
|
|
- Entry-ready strict code candidates: `{summary["counts"]["entry_ready_strict_code_rows"]}`.
|
- Latest-signal strict code candidates: `{summary["counts"]["latest_signal_strict_code_rows"]}`.
|
- Chart evidence files: `{summary["counts"]["chart_count"]}`.
|
|
## Boundary
|
|
`support_manual_decision` is intentionally left as `SUPPORT_REVIEW_REQUIRED`: the phrase
|
"bottom support must be strong" is an image/manual-review condition in the Wuji notes, not a
|
pure hard-code condition. This package prepares the strict code candidates and chart evidence
|
for manual support review and reviewer audit.
|
"""
|
|
|
def write_self_check(all_rows: pd.DataFrame, chart_rows: list[dict]) -> list[Path]:
|
checks = []
|
def add(check_id: str, ok: bool, detail: str) -> None:
|
checks.append({"check_id": check_id, "status": "PASS" if ok else "FAIL", "detail": detail})
|
|
add("NO_BJ_SYMBOL_RETAINED", int(all_rows["symbol"].str.endswith(".BJ").sum()) == 0, f"bj_rows={int(all_rows['symbol'].str.endswith('.BJ').sum())}")
|
add("STRICT_VOLUME_GE_2", bool(all_rows["volume_ratio"].ge(2.0).all()), f"min_volume_ratio={all_rows['volume_ratio'].min() if len(all_rows) else 'NA'}")
|
add("PRIOR_LIMITUP_30", bool(all_rows["prior_limitup_30_flag"].all()), f"rows={len(all_rows)}")
|
add("LONG_UPPER_SHADOW", bool((all_rows["upper_shadow_pct"].ge(3.0) & all_rows["upper_shadow_range_ratio"].ge(0.40)).all()), f"rows={len(all_rows)}")
|
add("PULLBACK_AFTER_LIMITUP", bool(all_rows["pullback_from_last_limitup_close_pct"].le(-3.0).all()), f"max_pullback_pct={all_rows['pullback_from_last_limitup_close_pct'].max() if len(all_rows) else 'NA'}")
|
add("PREV_HIGH_VOLUME_GUARD", bool(all_rows["prev_high_volume_pass_flag"].all()), f"rows={len(all_rows)}")
|
missing = [r["chart_path"] for r in chart_rows if not (ROOT / r["chart_path"]).exists()]
|
add("CHARTS_EXIST", len(missing) == 0, f"missing={len(missing)}; charts={len(chart_rows)}")
|
add("SUPPORT_REVIEW_NOT_AUTO_PASSED", set(all_rows["support_manual_decision"]) == {"SUPPORT_REVIEW_REQUIRED"} if len(all_rows) else True, "manual support review required")
|
|
items_path = ROOT / "self_check_items.csv"
|
with items_path.open("w", encoding="utf-8-sig", newline="") as f:
|
writer = csv.DictWriter(f, fieldnames=["check_id", "status", "detail"])
|
writer.writeheader()
|
writer.writerows(checks)
|
json_path = ROOT / "self_check.json"
|
json_path.write_text(
|
json.dumps(
|
{
|
"schema_version": "1.0",
|
"run_id": RUN_ID,
|
"status": "PASS" if all(c["status"] == "PASS" for c in checks) else "FAIL",
|
"items": checks,
|
},
|
ensure_ascii=False,
|
indent=2,
|
),
|
encoding="utf-8",
|
)
|
return [items_path, json_path]
|
|
|
def write_manifest(files: list[Path]) -> Path:
|
rows = []
|
for path in files:
|
rows.append(
|
{
|
"path": path.relative_to(ROOT).as_posix(),
|
"size": path.stat().st_size,
|
"sha256": sha256_file(path),
|
}
|
)
|
manifest = ROOT / "manifest.json"
|
manifest.write_text(
|
json.dumps(
|
{
|
"schema_version": "1.0",
|
"run_id": RUN_ID,
|
"generated_at": datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(timespec="seconds"),
|
"files": rows,
|
},
|
ensure_ascii=False,
|
indent=2,
|
),
|
encoding="utf-8",
|
)
|
return manifest
|
|
|
def main() -> None:
|
ROOT.mkdir(parents=True, exist_ok=True)
|
CHART_DIR.mkdir(parents=True, exist_ok=True)
|
source = load_source_module()
|
generated_at = datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(timespec="seconds")
|
with source.get_conn() as conn:
|
latest_date = pd.read_sql(
|
"SELECT MAX(trade_date) AS latest_trade_date FROM a_share_daily_price", conn
|
)["latest_trade_date"].iloc[0]
|
latest_date = pd.Timestamp(latest_date)
|
pull_start = (latest_date - pd.Timedelta(days=320)).strftime("%Y-%m-%d")
|
daily = pd.read_sql(
|
"""
|
SELECT trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount
|
FROM a_share_daily_price
|
WHERE trade_date BETWEEN %(start)s AND %(end)s
|
ORDER BY symbol, trade_date
|
""",
|
conn,
|
params={"start": pull_start, "end": latest_date.strftime("%Y-%m-%d")},
|
)
|
|
daily["trade_date"] = pd.to_datetime(daily["trade_date"])
|
for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]:
|
daily[col] = pd.to_numeric(daily[col], errors="coerce")
|
daily = add_features(daily)
|
breadth = breadth_from_daily(daily)
|
trade_dates = sorted(daily["trade_date"].drop_duplicates())
|
latest_signal_date = pd.Timestamp(trade_dates[-1])
|
entry_signal_date = pd.Timestamp(trade_dates[-2])
|
entry_date = latest_signal_date
|
|
entry_ready = build_strict_candidates(
|
daily, entry_signal_date, entry_date, "ENTRY_READY_STRICT", breadth
|
)
|
latest_signal = build_strict_candidates(
|
daily, latest_signal_date, None, "LATEST_SIGNAL_STRICT", breadth
|
)
|
all_rows = pd.concat([entry_ready, latest_signal], ignore_index=True)
|
|
out_cols = [
|
"candidate_id",
|
"scan_label",
|
"entry_trade_date",
|
"signal_trade_date",
|
"symbol",
|
"candidate_rank",
|
"strict_candidate_status",
|
"support_manual_decision",
|
"support_manual_reason_cn",
|
"market_gate_status",
|
"market_gate_open_flag",
|
"stock_count",
|
"up_count",
|
"flat_count",
|
"down_count",
|
"open_price",
|
"high_price",
|
"low_price",
|
"close_price",
|
"prev_close",
|
"daily_return_pct",
|
"volume",
|
"amount",
|
"prev5_avg_volume",
|
"prev10_avg_volume",
|
"volume_ratio",
|
"volume_ratio_ma10",
|
"upper_shadow_pct",
|
"upper_shadow_range_ratio",
|
"body_pct",
|
"prior_limitup_30_flag",
|
"last_prior_limitup_date",
|
"days_since_prior_limitup",
|
"last_prior_limitup_close",
|
"post_limitup_min_low",
|
"pullback_from_last_limitup_close_pct",
|
"touch_prev_high_flag",
|
"prev60_high",
|
"prev60_high_volume",
|
"prev60_high_ref_date",
|
"prev60_high_ref_policy",
|
"prev_high_volume_pass_flag",
|
"flat60_range_pct",
|
]
|
for df in [entry_ready, latest_signal, all_rows]:
|
for col in out_cols:
|
if col not in df.columns:
|
df[col] = pd.NA
|
|
chart_rows = []
|
for _, row in all_rows.iterrows():
|
hist = daily[
|
(daily["symbol"].eq(row["symbol"]))
|
& (daily["trade_date"].le(pd.Timestamp(row["signal_trade_date"])))
|
].copy()
|
chart_path = CHART_DIR / f"{row['candidate_id']}_{row['symbol']}.svg"
|
draw_chart(hist, row, chart_path)
|
chart_rows.append(
|
{
|
"candidate_id": row["candidate_id"],
|
"symbol": row["symbol"],
|
"scan_label": row["scan_label"],
|
"signal_trade_date": pd.Timestamp(row["signal_trade_date"]).strftime("%Y-%m-%d"),
|
"chart_path": chart_path.relative_to(ROOT).as_posix(),
|
"chart_exists": chart_path.exists(),
|
"chart_sha256": sha256_file(chart_path) if chart_path.exists() else "",
|
}
|
)
|
chart_audit = pd.DataFrame(chart_rows)
|
|
files = [
|
write_csv(all_rows[out_cols], "strict_candidate_ledger.csv"),
|
write_csv(entry_ready[out_cols], "entry_ready_strict_candidates.csv"),
|
write_csv(latest_signal[out_cols], "latest_signal_strict_candidates.csv"),
|
write_csv(chart_audit, "chart_evidence_audit.csv"),
|
]
|
|
summary = {
|
"schema_version": "1.0",
|
"run_id": RUN_ID,
|
"generated_at": generated_at,
|
"stage": "STRICT_DAILY_CANDIDATE_CODE_SCAN_SUPPORT_REVIEW_REQUIRED",
|
"latest_daily_trade_date": latest_signal_date.strftime("%Y-%m-%d"),
|
"rules": {
|
"exclude": "*.BJ",
|
"prior_limitup_30_flag": True,
|
"volume_ratio_prev5_min": 2.0,
|
"upper_shadow_pct_min": 3.0,
|
"upper_shadow_range_ratio_min": 0.40,
|
"pullback_from_last_limitup_close_pct_max": -3.0,
|
"prev_high_volume_pass_flag": True,
|
},
|
"counts": {
|
"entry_ready_strict_code_rows": int(len(entry_ready)),
|
"latest_signal_strict_code_rows": int(len(latest_signal)),
|
"all_strict_code_rows": int(len(all_rows)),
|
"chart_count": int(len(chart_rows)),
|
"bj_rows_retained": int(all_rows["symbol"].str.endswith(".BJ").sum()) if len(all_rows) else 0,
|
},
|
"market_gate": {
|
"entry_ready": {
|
"signal_trade_date": entry_signal_date.strftime("%Y-%m-%d"),
|
"entry_trade_date": entry_date.strftime("%Y-%m-%d"),
|
"market_gate_status": entry_ready["market_gate_status"].iloc[0] if len(entry_ready) else None,
|
"up_count": int(entry_ready["up_count"].iloc[0]) if len(entry_ready) else None,
|
},
|
"latest_signal": {
|
"signal_trade_date": latest_signal_date.strftime("%Y-%m-%d"),
|
"market_gate_status": latest_signal["market_gate_status"].iloc[0] if len(latest_signal) else None,
|
"up_count": int(latest_signal["up_count"].iloc[0]) if len(latest_signal) else None,
|
},
|
},
|
"boundary": [
|
"This is a strict daily code scan, not a buy recommendation.",
|
"Bottom support strength is not auto-passed; support_manual_decision remains SUPPORT_REVIEW_REQUIRED.",
|
"Minute-level buy point confirmation is not included.",
|
],
|
}
|
summary_json = ROOT / "summary.json"
|
summary_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8")
|
summary_md = ROOT / "summary.md"
|
summary_md.write_text(render_summary(summary), encoding="utf-8")
|
files.extend([summary_json, summary_md])
|
files.extend([ROOT / r["chart_path"] for r in chart_rows])
|
files.extend(write_self_check(all_rows, chart_rows))
|
files.append(Path(__file__))
|
manifest = write_manifest(files)
|
|
print(
|
json.dumps(
|
{
|
"run_id": RUN_ID,
|
"result_dir": ROOT.as_posix(),
|
"entry_ready_strict_code_rows": len(entry_ready),
|
"latest_signal_strict_code_rows": len(latest_signal),
|
"all_strict_code_rows": len(all_rows),
|
"chart_count": len(chart_rows),
|
"manifest": manifest.as_posix(),
|
},
|
ensure_ascii=False,
|
indent=2,
|
)
|
)
|
|
|
if __name__ == "__main__":
|
main()
|
