Tools/huheng-research
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: bootstrap housing research platform

Browse Source
This commit is contained in:
shay7sev
2026-06-23 09:43:56 +08:00
commit 479e3f16d4
34 changed files with 4339 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/shanghai_housing/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""Shanghai housing market research toolkit."""
__version__ = "0.1.0"

5
src/shanghai_housing/__main__.py Normal file
View File

@@ -0,0 +1,5 @@
from .cli import main
if __name__ == "__main__":
raise SystemExit(main())

70
src/shanghai_housing/cli.py Normal file
View File

@@ -0,0 +1,70 @@
from __future__ import annotations
import argparse
from pathlib import Path
from typing import Optional
from .db import DEFAULT_DB_PATH, connect, init_db, load_sample_data
from .indicators import compute_area_scores
from .reporting import format_scores_table, write_monthly_report
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="shanghai_housing",
description="Shanghai housing market research toolkit.",
)
parser.add_argument(
"--db",
type=Path,
default=DEFAULT_DB_PATH,
help=f"SQLite database path. Default: {DEFAULT_DB_PATH}",
)
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser("init-db", help="Create or update the SQLite schema.")
subparsers.add_parser("load-sample", help="Load demonstration sample data.")
score_parser = subparsers.add_parser("score", help="Print area scores for a month.")
score_parser.add_argument("--month", required=True, help="Month in YYYY-MM format.")
report_parser = subparsers.add_parser("report", help="Generate a monthly Markdown report.")
report_parser.add_argument("--month", required=True, help="Month in YYYY-MM format.")
report_parser.add_argument(
"--output",
type=Path,
help="Output Markdown path. Default: reports/monthly-YYYY-MM.md",
)
return parser
def main(argv: Optional[list[str]] = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
if args.command == "init-db":
path = init_db(args.db)
print(f"Initialized database: {path}")
return 0
if args.command == "load-sample":
load_sample_data(args.db)
print(f"Loaded sample data into: {args.db}")
return 0
if args.command == "score":
with connect(args.db) as conn:
scores = compute_area_scores(conn, args.month)
print(format_scores_table(scores))
return 0
if args.command == "report":
output = args.output or Path("reports") / f"monthly-{args.month}.md"
with connect(args.db) as conn:
path = write_monthly_report(conn, args.month, output)
print(f"Generated report: {path}")
return 0
parser.error(f"Unknown command: {args.command}")
return 2

109
src/shanghai_housing/db.py Normal file
View File

@@ -0,0 +1,109 @@
from __future__ import annotations
import csv
import sqlite3
from pathlib import Path
from typing import Iterable, Mapping, Union
PROJECT_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_DB_PATH = PROJECT_ROOT / "data" / "shanghai_housing.sqlite"
SCHEMA_PATH = Path(__file__).with_name("schema.sql")
SAMPLE_DIR = PROJECT_ROOT / "data" / "sample"
def connect(db_path: Union[Path, str] = DEFAULT_DB_PATH) -> sqlite3.Connection:
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
return conn
def init_db(db_path: Union[Path, str] = DEFAULT_DB_PATH) -> Path:
path = Path(db_path)
path.parent.mkdir(parents=True, exist_ok=True)
with connect(path) as conn:
conn.executescript(SCHEMA_PATH.read_text(encoding="utf-8"))
return path
def load_sample_data(db_path: Union[Path, str] = DEFAULT_DB_PATH) -> None:
init_db(db_path)
with connect(db_path) as conn:
clear_tables(
conn,
[
"neighborhood_monthly_metrics",
"neighborhoods",
"area_monthly_metrics",
"policy_events",
"areas",
],
)
load_csv(conn, "areas", SAMPLE_DIR / "areas.csv")
load_csv(conn, "area_monthly_metrics", SAMPLE_DIR / "area_monthly_metrics.csv")
load_csv(conn, "neighborhoods", SAMPLE_DIR / "neighborhoods.csv")
load_csv(
conn,
"neighborhood_monthly_metrics",
SAMPLE_DIR / "neighborhood_monthly_metrics.csv",
)
def clear_tables(conn: sqlite3.Connection, tables: Iterable[str]) -> None:
for table in tables:
conn.execute(f"DELETE FROM {table}")
def load_csv(conn: sqlite3.Connection, table: str, path: Path) -> int:
with path.open("r", encoding="utf-8-sig", newline="") as handle:
rows = list(csv.DictReader(handle))
if not rows:
return 0
columns = list(rows[0].keys())
placeholders = ", ".join(["?"] * len(columns))
column_sql = ", ".join(columns)
update_sql = ", ".join([f"{column}=excluded.{column}" for column in columns])
sql = (
f"INSERT INTO {table} ({column_sql}) VALUES ({placeholders}) "
f"ON CONFLICT DO UPDATE SET {update_sql}"
)
conn.executemany(sql, [tuple(row[column] for column in columns) for row in rows])
return len(rows)
def fetch_area_metrics(conn: sqlite3.Connection, month: str) -> list[sqlite3.Row]:
return conn.execute(
"""
SELECT
a.area_id,
a.name,
a.district,
a.segment,
m.*
FROM area_monthly_metrics m
JOIN areas a ON a.area_id = m.area_id
WHERE m.month = ?
ORDER BY a.district, a.name
""",
(month,),
).fetchall()
def fetch_area_history(
conn: sqlite3.Connection, area_id: str, through_month: str
) -> list[sqlite3.Row]:
return conn.execute(
"""
SELECT *
FROM area_monthly_metrics
WHERE area_id = ? AND month <= ?
ORDER BY month
""",
(area_id, through_month),
).fetchall()
def row_to_dict(row: Union[sqlite3.Row, Mapping[str, object]]) -> dict[str, object]:
return {key: row[key] for key in row.keys()} if isinstance(row, sqlite3.Row) else dict(row)

212
src/shanghai_housing/indicators.py Normal file
View File

@@ -0,0 +1,212 @@
from __future__ import annotations
import sqlite3
from dataclasses import asdict, dataclass
from datetime import date
from typing import Optional
from .db import fetch_area_history, fetch_area_metrics
@dataclass(frozen=True)
class AreaScore:
area_id: str
name: str
district: str
segment: str
month: str
investment_score: float
recommendation: str
liquidity_score: float
momentum_score: float
rent_support_score: float
safety_margin_score: float
credit_support_score: float
supply_risk_score: float
valuation_pressure_score: float
transaction_count: int
transaction_price_psm: float
listing_count: int
listing_price_psm: float
rent_price_psm: float
median_days_on_market: float
annual_rent_yield_pct: float
listing_pressure_ratio: float
discount_pct: float
price_momentum_pct: float
volume_momentum_pct: float
def to_dict(self) -> dict[str, object]:
return asdict(self)
def previous_month(month: str) -> str:
year, month_num = [int(part) for part in month.split("-")]
current = date(year, month_num, 1)
if current.month == 1:
return f"{current.year - 1}-12"
return f"{current.year}-{current.month - 1:02d}"
def compute_area_scores(conn: sqlite3.Connection, month: str) -> list[AreaScore]:
current_rows = fetch_area_metrics(conn, month)
if not current_rows:
raise ValueError(f"No area metrics found for month {month}.")
prior_month = previous_month(month)
prior_rows = {row["area_id"]: row for row in fetch_area_metrics(conn, prior_month)}
max_transactions = max(float(row["transaction_count"]) for row in current_rows) or 1.0
scores = [
score_area(row, prior_rows.get(row["area_id"]), max_transactions, conn)
for row in current_rows
]
return sorted(scores, key=lambda score: score.investment_score, reverse=True)
def score_area(
row: sqlite3.Row,
prior_row: Optional[sqlite3.Row],
max_transactions: float,
conn: sqlite3.Connection,
) -> AreaScore:
transaction_count = int(row["transaction_count"])
transaction_price = float(row["transaction_price_psm"])
listing_count = int(row["listing_count"])
listing_price = float(row["listing_price_psm"])
rent_price = float(row["rent_price_psm"])
days_on_market = float(row["median_days_on_market"])
new_supply_units = int(row["new_supply_units"])
mortgage_rate = float(row["mortgage_rate_pct"])
policy_signal = int(row["policy_signal"])
annual_rent_yield = safe_div(rent_price * 12, transaction_price) * 100
listing_pressure = safe_div(listing_count, transaction_count)
supply_ratio = safe_div(new_supply_units, transaction_count)
discount_pct = safe_div(listing_price - transaction_price, listing_price) * 100
price_momentum = momentum_pct(
transaction_price,
float(prior_row["transaction_price_psm"]) if prior_row else None,
)
volume_momentum = momentum_pct(
transaction_count,
int(prior_row["transaction_count"]) if prior_row else None,
)
history = fetch_area_history(conn, row["area_id"], row["month"])
price_percentile = historical_percentile(
transaction_price,
[float(item["transaction_price_psm"]) for item in history],
)
liquidity_score = clamp(
0.65 * (transaction_count / max_transactions * 100)
+ 0.35 * low_better(days_on_market, good=45, bad=120)
)
momentum_score = clamp(
0.55 * high_better(volume_momentum, bad=-20, good=25)
+ 0.45 * high_better(price_momentum, bad=-3, good=4)
)
rent_support_score = high_better(annual_rent_yield, bad=1.0, good=2.4)
supply_risk_score = clamp(
0.60 * high_better(listing_pressure, bad=3.5, good=9.0)
+ 0.40 * high_better(supply_ratio, bad=0.5, good=4.0)
)
valuation_pressure_score = clamp(
0.60 * price_percentile + 0.40 * (100 - rent_support_score)
)
safety_margin_score = clamp(
0.55 * high_better(discount_pct, bad=1.5, good=8.0)
+ 0.45 * (100 - valuation_pressure_score)
)
credit_support_score = clamp(
0.70 * low_better(mortgage_rate, good=3.2, bad=5.0)
+ 0.30 * ((policy_signal + 2) / 4 * 100)
)
investment_score = clamp(
0.25 * liquidity_score
+ 0.20 * momentum_score
+ 0.20 * rent_support_score
+ 0.15 * safety_margin_score
+ 0.10 * credit_support_score
+ 0.10 * (100 - supply_risk_score)
)
return AreaScore(
area_id=row["area_id"],
name=row["name"],
district=row["district"],
segment=row["segment"],
month=row["month"],
investment_score=round(investment_score, 1),
recommendation=recommendation(investment_score),
liquidity_score=round(liquidity_score, 1),
momentum_score=round(momentum_score, 1),
rent_support_score=round(rent_support_score, 1),
safety_margin_score=round(safety_margin_score, 1),
credit_support_score=round(credit_support_score, 1),
supply_risk_score=round(supply_risk_score, 1),
valuation_pressure_score=round(valuation_pressure_score, 1),
transaction_count=transaction_count,
transaction_price_psm=round(transaction_price, 1),
listing_count=listing_count,
listing_price_psm=round(listing_price, 1),
rent_price_psm=round(rent_price, 1),
median_days_on_market=round(days_on_market, 1),
annual_rent_yield_pct=round(annual_rent_yield, 2),
listing_pressure_ratio=round(listing_pressure, 2),
discount_pct=round(discount_pct, 2),
price_momentum_pct=round(price_momentum, 2),
volume_momentum_pct=round(volume_momentum, 2),
)
def safe_div(numerator: float, denominator: float) -> float:
return numerator / denominator if denominator else 0.0
def momentum_pct(current: float, previous: Optional[float]) -> float:
if previous is None or previous == 0:
return 0.0
return (current - previous) / previous * 100
def historical_percentile(current: float, values: list[float]) -> float:
if not values:
return 50.0
low = min(values)
high = max(values)
if high == low:
return 50.0
return clamp((current - low) / (high - low) * 100)
def high_better(value: float, bad: float, good: float) -> float:
if value <= bad:
return 0.0
if value >= good:
return 100.0
return clamp((value - bad) / (good - bad) * 100)
def low_better(value: float, good: float, bad: float) -> float:
if value <= good:
return 100.0
if value >= bad:
return 0.0
return clamp((bad - value) / (bad - good) * 100)
def clamp(value: float, lower: float = 0.0, upper: float = 100.0) -> float:
return max(lower, min(upper, value))
def recommendation(score: float) -> str:
if score >= 75:
return "重点研究"
if score >= 65:
return "观察池"
if score >= 50:
return "中性观望"
return "谨慎等待"

131
src/shanghai_housing/reporting.py Normal file
View File

@@ -0,0 +1,131 @@
from __future__ import annotations
import sqlite3
from pathlib import Path
from .indicators import AreaScore, compute_area_scores
def build_monthly_report(conn: sqlite3.Connection, month: str) -> str:
scores = compute_area_scores(conn, month)
top = scores[0]
weakest = scores[-1]
high_supply_risk = max(scores, key=lambda item: item.supply_risk_score)
avg_score = sum(score.investment_score for score in scores) / len(scores)
avg_yield = sum(score.annual_rent_yield_pct for score in scores) / len(scores)
lines = [
f"# 上海房市投资研究月报 {month}",
"",
"> 本报告由本地研究系统生成。当前数据若来自 `data/sample`，仅用于演示方法，不构成真实投资建议。",
"",
"## 核心结论",
"",
f"- 综合评分最高板块：{top.name}（{top.investment_score}，{top.recommendation}）。",
f"- 综合评分最低板块：{weakest.name}（{weakest.investment_score}，{weakest.recommendation}）。",
f"- 供应压力最高板块：{high_supply_risk.name}（供应风险 {high_supply_risk.supply_risk_score}）。",
f"- 样本平均投资观察评分：{avg_score:.1f}；样本平均年化租金收益率：{avg_yield:.2f}%。",
"",
"## 板块评分",
"",
"| 排名 | 板块 | 行政区 | 类型 | 综合分 | 建议 | 流动性 | 动量 | 租金支撑 | 安全边际 | 供应风险 |",
"| --- | --- | --- | --- | ---: | --- | ---: | ---: | ---: | ---: | ---: |",
]
for index, score in enumerate(scores, start=1):
lines.append(
"| {rank} | {name} | {district} | {segment} | {investment_score:.1f} | "
"{recommendation} | {liquidity_score:.1f} | {momentum_score:.1f} | "
"{rent_support_score:.1f} | {safety_margin_score:.1f} | {supply_risk_score:.1f} |".format(
rank=index,
name=score.name,
district=score.district,
segment=score.segment,
investment_score=score.investment_score,
recommendation=score.recommendation,
liquidity_score=score.liquidity_score,
momentum_score=score.momentum_score,
rent_support_score=score.rent_support_score,
safety_margin_score=score.safety_margin_score,
supply_risk_score=score.supply_risk_score,
)
)
lines.extend(
[
"",
"## 关键指标明细",
"",
"| 板块 | 成交套数 | 成交均价/㎡ | 挂牌套数 | 挂牌均价/㎡ | 去化压力 | 成交周期 | 年化租金收益率 | 成交折价 | 价格动量 | 成交动量 |",
"| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
]
)
for score in scores:
lines.append(
"| {name} | {transaction_count} | {transaction_price_psm:,.0f} | {listing_count} | "
"{listing_price_psm:,.0f} | {listing_pressure_ratio:.2f} | "
"{median_days_on_market:.0f}天 | {annual_rent_yield_pct:.2f}% | "
"{discount_pct:.2f}% | {price_momentum_pct:.2f}% | {volume_momentum_pct:.2f}% |".format(
**score.to_dict()
)
)
lines.extend(
[
"",
"## 模型口径",
"",
"- 综合分由流动性、成交/价格动量、租金支撑、安全边际、信贷政策环境、供应风险共同决定。",
"- 供应风险越高表示挂牌库存、新增供应相对成交越重；综合分会对其反向处理。",
"- 安全边际主要来自成交/挂牌折价和历史价格分位。",
"- 第一版模型适合建立观察池，不适合单独作为买卖决策。",
"",
"## 下一步人工复核",
"",
"- 检查高分板块内的小区分化，剔除楼龄、物业、噪音、硬伤户型等不可量化风险。",
"- 对供应风险高的板块追踪新房开盘节奏、竞品库存和开发商价格策略。",
"- 将政策事件和信贷变化作为情景变量，而不是简单线性外推。",
"",
]
)
return "\n".join(lines)
def write_monthly_report(
conn: sqlite3.Connection, month: str, output_path: Path
) -> Path:
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(build_monthly_report(conn, month), encoding="utf-8")
return output_path
def format_scores_table(scores: list[AreaScore]) -> str:
rows = [
(
score.name,
f"{score.investment_score:.1f}",
score.recommendation,
f"{score.liquidity_score:.1f}",
f"{score.momentum_score:.1f}",
f"{score.rent_support_score:.1f}",
f"{score.supply_risk_score:.1f}",
)
for score in scores
]
header = ("板块", "综合分", "建议", "流动性", "动量", "租金", "供应风险")
return render_plain_table(header, rows)
def render_plain_table(header: tuple[str, ...], rows: list[tuple[str, ...]]) -> str:
widths = [
max(len(str(row[index])) for row in [header, *rows])
for index in range(len(header))
]
lines = [format_table_row(header, widths), format_table_row(tuple("-" * width for width in widths), widths)]
lines.extend(format_table_row(row, widths) for row in rows)
return "\n".join(lines)
def format_table_row(row: tuple[str, ...], widths: list[int]) -> str:
return " ".join(str(value).ljust(widths[index]) for index, value in enumerate(row))

68
src/shanghai_housing/schema.sql Normal file
View File

@@ -0,0 +1,68 @@
PRAGMA foreign_keys = ON;
CREATE TABLE IF NOT EXISTS areas (
area_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
district TEXT NOT NULL,
segment TEXT NOT NULL,
notes TEXT DEFAULT ''
);
CREATE TABLE IF NOT EXISTS area_monthly_metrics (
area_id TEXT NOT NULL,
month TEXT NOT NULL,
transaction_count INTEGER NOT NULL CHECK (transaction_count >= 0),
transaction_price_psm REAL NOT NULL CHECK (transaction_price_psm >= 0),
listing_count INTEGER NOT NULL CHECK (listing_count >= 0),
listing_price_psm REAL NOT NULL CHECK (listing_price_psm >= 0),
median_days_on_market REAL NOT NULL CHECK (median_days_on_market >= 0),
rent_price_psm REAL NOT NULL CHECK (rent_price_psm >= 0),
new_supply_units INTEGER NOT NULL CHECK (new_supply_units >= 0),
land_residential_gfa_sqm REAL NOT NULL CHECK (land_residential_gfa_sqm >= 0),
mortgage_rate_pct REAL NOT NULL CHECK (mortgage_rate_pct >= 0),
policy_signal INTEGER NOT NULL DEFAULT 0 CHECK (policy_signal BETWEEN -2 AND 2),
source TEXT NOT NULL DEFAULT 'manual',
PRIMARY KEY (area_id, month),
FOREIGN KEY (area_id) REFERENCES areas(area_id)
);
CREATE TABLE IF NOT EXISTS neighborhoods (
neighborhood_id TEXT PRIMARY KEY,
area_id TEXT NOT NULL,
name TEXT NOT NULL,
built_year INTEGER,
property_type TEXT NOT NULL,
metro_distance_m INTEGER,
school_quality TEXT,
notes TEXT DEFAULT '',
FOREIGN KEY (area_id) REFERENCES areas(area_id)
);
CREATE TABLE IF NOT EXISTS neighborhood_monthly_metrics (
neighborhood_id TEXT NOT NULL,
month TEXT NOT NULL,
transaction_count INTEGER NOT NULL CHECK (transaction_count >= 0),
transaction_price_psm REAL NOT NULL CHECK (transaction_price_psm >= 0),
listing_count INTEGER NOT NULL CHECK (listing_count >= 0),
listing_price_psm REAL NOT NULL CHECK (listing_price_psm >= 0),
rent_price_psm REAL NOT NULL CHECK (rent_price_psm >= 0),
median_days_on_market REAL NOT NULL CHECK (median_days_on_market >= 0),
available_units INTEGER NOT NULL CHECK (available_units >= 0),
source TEXT NOT NULL DEFAULT 'manual',
PRIMARY KEY (neighborhood_id, month),
FOREIGN KEY (neighborhood_id) REFERENCES neighborhoods(neighborhood_id)
);
CREATE TABLE IF NOT EXISTS policy_events (
event_id INTEGER PRIMARY KEY AUTOINCREMENT,
event_date TEXT NOT NULL,
level TEXT NOT NULL,
title TEXT NOT NULL,
impact_direction INTEGER NOT NULL CHECK (impact_direction BETWEEN -2 AND 2),
notes TEXT DEFAULT '',
source_url TEXT
);
CREATE INDEX IF NOT EXISTS idx_area_metrics_month ON area_monthly_metrics(month);
CREATE INDEX IF NOT EXISTS idx_neighborhood_metrics_month ON neighborhood_monthly_metrics(month);
CREATE INDEX IF NOT EXISTS idx_policy_events_date ON policy_events(event_date);