#!/usr/bin/env python3
"""
Generate trend analysis HTML from JD法拍 CSV files.
Reads all *_法拍房源*.csv files in output/法拍/, merges by paimaiId,
computes monthly trends (上架量/流拍量), and generates an interactive HTML chart.
"""
import csv
import json
import os
import glob
from collections import Counter
CSV_DIR = "output/法拍"
OUTPUT_HTML = os.path.join(CSV_DIR, "东莞法拍房趋势分析.html")
def read_all_csvs(csv_dir):
all_items = {}
csv_files = glob.glob(os.path.join(csv_dir, "*_法拍房源*.csv"))
for csv_path in csv_files:
try:
with open(csv_path, encoding="utf-8-sig") as f:
reader = csv.DictReader(f)
for row in reader:
pid = row.get("拍卖ID", "")
if not pid:
continue
if pid not in all_items:
all_items[pid] = row
else:
existing = all_items[pid]
if row.get("结束时间") and not existing.get("结束时间"):
all_items[pid] = row
except Exception as e:
print(f"Warning: failed to read {csv_path}: {e}")
return list(all_items.values())
def compute_monthly_trends(items, region_filter=None):
if region_filter:
items = [r for r in items if region_filter in r.get("标题", "")]
listed_by_month = Counter()
failed_by_month = Counter()
ended_by_month = Counter()
for r in items:
start = r.get("开始时间", "")[:7]
end = r.get("结束时间", "")[:7]
status = r.get("状态", "")
bid_count = r.get("出价次数", "")
if start:
listed_by_month[start] += 1
if end and status == "已结束":
ended_by_month[end] += 1
if bid_count in ("0", ""):
failed_by_month[end] += 1
month_set = (
set(listed_by_month.keys())
| set(ended_by_month.keys())
| set(failed_by_month.keys())
)
if not month_set:
return []
all_months = sorted(month_set)
start_m = all_months[0]
end_m = all_months[-1]
full_months = []
y, m = int(start_m[:4]), int(start_m[5:7])
ey, em = int(end_m[:4]), int(end_m[5:7])
while (y, m) <= (ey, em):
full_months.append(f"{y:04d}-{m:02d}")
m += 1
if m > 12:
m = 1
y += 1
results = []
for month in full_months:
listed = listed_by_month.get(month, 0)
failed = failed_by_month.get(month, 0)
ended = ended_by_month.get(month, 0)
rate = round(failed / ended * 100, 1) if ended > 0 else None
results.append(
{
"month": month,
"listed": listed,
"failed": failed,
"ended": ended,
"rate": rate,
}
)
return results
def generate_svg(data, panel_idx):
n = len(data)
if n == 0:
return ""
W, H = 1120, 240
padL, padR, padT, padB = 52, 24, 16, 40
plotW = W - padL - padR
plotH = H - padT - padB
max_val = max((max(d["listed"], d["failed"]) for d in data), default=1)
max_val = max(max_val, 5)
y_ticks = 5
y_step_val = max_val / y_ticks if max_val > 0 else 1
def y_pos(val):
if max_val == 0:
return padT + plotH
return padT + plotH - (val / max_val) * plotH
def x_pos(i):
if n == 1:
return padL + plotW / 2
return padL + (i / (n - 1)) * plotW
svg_parts = []
# Y-axis grid lines and labels
for t in range(y_ticks + 1):
val = t * y_step_val
y = y_pos(val)
svg_parts.append(
f''
)
svg_parts.append(
f'{int(val)}'
)
# X-axis labels (year markers)
prev_year = None
for i, d in enumerate(data):
year = d["month"][:4]
if year != prev_year:
x = x_pos(i)
svg_parts.append(
f'{year}'
)
if prev_year is not None:
svg_parts.append(
f''
)
prev_year = year
# Axis line
svg_parts.append(
f''
)
# Area fill for listed (blue)
area_pts = [f"{x_pos(0):.1f},{y_pos(data[0]['listed']):.1f}"]
for i, d in enumerate(data):
area_pts.append(f"{x_pos(i):.1f},{y_pos(d['listed']):.1f}")
area_pts.append(f"{x_pos(n-1):.1f},{padT+plotH:.1f}")
area_pts.append(f"{x_pos(0):.1f},{padT+plotH:.1f}")
svg_parts.append(
f''
)
# Area fill for failed (orange)
area_pts2 = [f"{x_pos(0):.1f},{y_pos(data[0]['failed']):.1f}"]
for i, d in enumerate(data):
area_pts2.append(f"{x_pos(i):.1f},{y_pos(d['failed']):.1f}")
area_pts2.append(f"{x_pos(n-1):.1f},{padT+plotH:.1f}")
area_pts2.append(f"{x_pos(0):.1f},{padT+plotH:.1f}")
svg_parts.append(
f''
)
# Line for listed
line_pts = [f"{x_pos(i):.1f},{y_pos(d['listed']):.1f}" for i, d in enumerate(data)]
svg_parts.append(
f''
)
# Line for failed
line_pts2 = [f"{x_pos(i):.1f},{y_pos(d['failed']):.1f}" for i, d in enumerate(data)]
svg_parts.append(
f''
)
# Data point circles + hit areas
hit_w = plotW / n
for i, d in enumerate(data):
x = x_pos(i)
y1 = y_pos(d["listed"])
y2 = y_pos(d["failed"])
r = 2.5 if n > 50 else 3.5
svg_parts.append(
f''
)
svg_parts.append(
f''
)
hx = x - hit_w / 2
svg_parts.append(
f''
)
# Last point data labels
last = data[-1]
lx = x_pos(n - 1)
svg_parts.append(
f'{last["listed"]}'
)
if last["failed"] > 0:
svg_parts.append(
f'{last["failed"]}'
)
# Crosshair
svg_parts.append(
f''
)
return (
f'"
)
def generate_table_row_html(region, data):
rows = [f'
{region}（共 {len(data)} 个月）
']
for d in data:
rate_str = f'{d["rate"]:.1f}%' if d["rate"] is not None else "—"
rows.append(
f'
{d["month"]}
'
f'
{d["listed"]}
{d["failed"]}
'
f'
{d["ended"]}
{rate_str}
'
)
return "\n".join(rows)
def generate_html(panels, items_count):
svgs = {}
for key, data in panels.items():
idx = list(panels.keys()).index(key)
svgs[key] = generate_svg(data, idx)
data_json = json.dumps(panels, ensure_ascii=False)
panel_keys = list(panels.keys())
# Date range
all_months = []
for data in panels.values():
all_months.extend([d["month"] for d in data])
date_range = f"{min(all_months)} – {max(all_months)}" if all_months else ""
# Sample counts
sample_counts = {}
for key in panel_keys:
if key == "东莞全市":
sample_counts[key] = items_count
else:
sample_counts[key] = sum(
1 for d in panels[key] for _ in range(d["listed"])
)
table_html = "\n".join(
generate_table_row_html(key, data) for key, data in panels.items()
)
panel_html = ""
for i, key in enumerate(panel_keys):
subtitle = f"样本量 {sample_counts[key]} 套" if key != "东莞全市" else f"全市法拍住宅 · 共 {items_count} 条记录"
panel_html += f"""