feat: add scenario analysis workspace

This commit is contained in:
2026-06-24 16:43:27 +08:00
parent b559df50d8
commit f0ac47e040
7 changed files with 986 additions and 20 deletions

View File

@@ -0,0 +1,18 @@
CREATE TABLE IF NOT EXISTS app.scenario_runs (
scenario_run_id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
target_month TEXT NOT NULL CHECK (target_month ~ '^[0-9]{4}-[0-9]{2}$'),
area_id TEXT REFERENCES silver.areas(area_id),
interest_rate_delta_bps DOUBLE PRECISION NOT NULL DEFAULT 0,
policy_signal_delta INTEGER NOT NULL DEFAULT 0 CHECK (policy_signal_delta BETWEEN -4 AND 4),
supply_delta_pct DOUBLE PRECISION NOT NULL DEFAULT 0 CHECK (supply_delta_pct BETWEEN -100 AND 500),
rent_delta_pct DOUBLE PRECISION NOT NULL DEFAULT 0 CHECK (rent_delta_pct BETWEEN -80 AND 200),
notes TEXT NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS idx_scenario_runs_month
ON app.scenario_runs(target_month, scenario_run_id DESC);
CREATE INDEX IF NOT EXISTS idx_scenario_runs_area
ON app.scenario_runs(area_id, scenario_run_id DESC);

View File

@@ -9,14 +9,16 @@ use crate::models::{
AreaProfile, AreaScore, AreaScoreLineage, AreaScoreLineageQuery, AreaScoreModelRunResponse,
AreaScoreRunDiff, AreaScoreRunDiffItem, AreaScoreSummary, CompareQuery, ComparisonMetric,
ComparisonMetricValue, CreateAreaScoreModelRun, CreateDataSource, CreateIngestionRun,
CreateRawArtifact, CreateWatchlistEvent, CreateWatchlistItem, DataSource, DiagnosticMetric,
FinishIngestionRun, ImportExecutionRequest, ImportExecutionResponse, ImportValidationRequest,
ImportValidationResponse, IngestionRun, IngestionRunQuery, MarketOverview, ModelRun,
ModelRunDiffQuery, ModelRunQuery, MonthQuery, Neighborhood, NeighborhoodComparison,
NeighborhoodComparisonItem, NeighborhoodDetail, NeighborhoodMonthlyMetric, NeighborhoodQuery,
NeighborhoodScore, NeighborhoodScoreQuery, RawArtifact, RawArtifactQuery, SimilarNeighborhood,
SimilarNeighborhoodCandidate, SimilarNeighborhoodQuery, SimilarNeighborhoodResponse,
UpdateWatchlistItem, WatchlistEvent, WatchlistItem, WatchlistQuery,
CreateRawArtifact, CreateScenarioRun, CreateWatchlistEvent, CreateWatchlistItem, DataSource,
DiagnosticMetric, FinishIngestionRun, ImportExecutionRequest, ImportExecutionResponse,
ImportValidationRequest, ImportValidationResponse, IngestionRun, IngestionRunQuery,
MarketOverview, ModelRun, ModelRunDiffQuery, ModelRunQuery, MonthQuery, Neighborhood,
NeighborhoodComparison, NeighborhoodComparisonItem, NeighborhoodDetail,
NeighborhoodMonthlyMetric, NeighborhoodQuery, NeighborhoodScore, NeighborhoodScoreQuery,
RawArtifact, RawArtifactQuery, ScenarioAreaResult, ScenarioRun, ScenarioRunQuery,
ScenarioRunResponse, ScenarioVariant, SimilarNeighborhood, SimilarNeighborhoodCandidate,
SimilarNeighborhoodQuery, SimilarNeighborhoodResponse, UpdateWatchlistItem, WatchlistEvent,
WatchlistItem, WatchlistQuery,
};
use crate::scoring::{compute_area_scores, previous_month, AreaMetricRow, ComputedAreaScore};
use crate::state::AppState;
@@ -1749,6 +1751,103 @@ pub async fn create_watchlist_event(
Ok(Json(event))
}
pub async fn list_scenario_runs(
State(state): State<AppState>,
Query(query): Query<ScenarioRunQuery>,
) -> ApiResult<Json<Vec<ScenarioRun>>> {
if let Some(month) = query.target_month.as_deref() {
validate_month(month)?;
}
if let Some(area_id) = query.area_id.as_deref() {
validate_identifier(area_id)?;
}
let runs = sqlx::query_as::<_, ScenarioRun>(
r#"
SELECT
scenario_run_id,
name,
target_month,
area_id,
interest_rate_delta_bps,
policy_signal_delta,
supply_delta_pct,
rent_delta_pct,
notes,
created_at::text AS created_at
FROM app.scenario_runs
WHERE ($1::text IS NULL OR target_month = $1)
AND ($2::text IS NULL OR area_id = $2)
ORDER BY scenario_run_id DESC
LIMIT 50
"#,
)
.bind(query.target_month)
.bind(query.area_id)
.fetch_all(&state.pool)
.await?;
Ok(Json(runs))
}
pub async fn create_scenario_run(
State(state): State<AppState>,
Json(payload): Json<CreateScenarioRun>,
) -> ApiResult<Json<ScenarioRunResponse>> {
validate_month(&payload.target_month)?;
payload
.validate()
.map_err(|message| ApiError::BadRequest(message.to_string()))?;
if let Some(area_id) = payload.area_id.as_deref() {
validate_identifier(area_id)?;
}
let mut tx = state.pool.begin().await?;
let run = sqlx::query_as::<_, ScenarioRun>(
r#"
INSERT INTO app.scenario_runs (
name,
target_month,
area_id,
interest_rate_delta_bps,
policy_signal_delta,
supply_delta_pct,
rent_delta_pct,
notes
)
VALUES ($1, $2, $3, $4, $5, $6, $7, COALESCE($8, ''))
RETURNING
scenario_run_id,
name,
target_month,
area_id,
interest_rate_delta_bps,
policy_signal_delta,
supply_delta_pct,
rent_delta_pct,
notes,
created_at::text AS created_at
"#,
)
.bind(payload.name.trim().to_string())
.bind(payload.target_month.trim().to_string())
.bind(payload.area_id.clone())
.bind(payload.interest_rate_delta_bps)
.bind(payload.policy_signal_delta)
.bind(payload.supply_delta_pct)
.bind(payload.rent_delta_pct)
.bind(payload.notes.clone())
.fetch_one(&mut *tx)
.await?;
let variants = build_scenario_variants(&mut tx, &run)
.await
.map_err(ApiError::BadRequest)?;
tx.commit().await?;
Ok(Json(ScenarioRunResponse { run, variants }))
}
async fn insert_watchlist_event(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
watchlist_item_id: i64,
@@ -1812,6 +1911,252 @@ async fn fetch_area_scores(state: &AppState, month: &str) -> Result<Vec<AreaScor
.await
}
async fn build_scenario_variants(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
run: &ScenarioRun,
) -> Result<Vec<ScenarioVariant>, String> {
let current_rows = fetch_area_metric_rows(tx, &run.target_month)
.await
.map_err(|error| format!("failed to fetch current area metrics: {error}"))?;
if current_rows.is_empty() {
return Err(format!(
"no area metrics found for month {}",
run.target_month
));
}
let target_rows = current_rows
.iter()
.filter(|row| {
run.area_id
.as_deref()
.map(|area_id| row.area_id == area_id)
.unwrap_or(true)
})
.cloned()
.collect::<Vec<_>>();
if target_rows.is_empty() {
return Err("no area metrics found for scenario area".to_string());
}
let prior_month =
previous_month(&run.target_month).ok_or_else(|| "invalid month".to_string())?;
let prior_rows = fetch_area_metric_rows(tx, &prior_month)
.await
.map_err(|error| format!("failed to fetch prior area metrics: {error}"))?;
let price_history = fetch_area_price_history(tx, &run.target_month)
.await
.map_err(|error| format!("failed to fetch area price history: {error}"))?;
let baseline_scores = compute_area_scores(&current_rows, &prior_rows, &price_history);
Ok(vec![
build_scenario_variant(
"downside",
"谨慎",
1.35,
&target_rows,
&prior_rows,
&price_history,
&baseline_scores,
run,
),
build_scenario_variant(
"base",
"基准",
1.0,
&target_rows,
&prior_rows,
&price_history,
&baseline_scores,
run,
),
build_scenario_variant(
"upside",
"乐观",
0.65,
&target_rows,
&prior_rows,
&price_history,
&baseline_scores,
run,
),
])
}
fn build_scenario_variant(
key: &str,
name: &str,
multiplier: f64,
rows: &[AreaMetricRow],
prior_rows: &[AreaMetricRow],
price_history: &std::collections::HashMap<String, Vec<f64>>,
baseline_scores: &[ComputedAreaScore],
run: &ScenarioRun,
) -> ScenarioVariant {
let adjusted_rows = rows
.iter()
.map(|row| adjusted_area_metric_row(row, run, multiplier))
.collect::<Vec<_>>();
let projected_scores = compute_area_scores(&adjusted_rows, prior_rows, price_history);
let baseline_by_area = baseline_scores
.iter()
.map(|score| (score.area_id.as_str(), score))
.collect::<std::collections::HashMap<_, _>>();
let adjusted_by_area = adjusted_rows
.iter()
.map(|row| (row.area_id.as_str(), row))
.collect::<std::collections::HashMap<_, _>>();
let items = projected_scores
.into_iter()
.filter_map(|projected| {
let baseline = baseline_by_area.get(projected.area_id.as_str())?;
let adjusted = adjusted_by_area.get(projected.area_id.as_str())?;
Some(scenario_area_result(
baseline, &projected, adjusted, run, multiplier,
))
})
.collect::<Vec<_>>();
ScenarioVariant {
key: key.to_string(),
name: name.to_string(),
assumptions: scenario_assumptions(run, multiplier),
items,
}
}
fn adjusted_area_metric_row(
row: &AreaMetricRow,
run: &ScenarioRun,
multiplier: f64,
) -> AreaMetricRow {
let price_delta_pct = scenario_price_delta_pct(run, multiplier);
let transaction_delta_pct = scenario_transaction_delta_pct(run, multiplier);
let listing_delta_pct = (run.supply_delta_pct * multiplier
+ run.interest_rate_delta_bps.max(0.0) * 0.08
- (run.policy_signal_delta.max(0) as f64) * 3.0)
.clamp(-60.0, 180.0);
AreaMetricRow {
transaction_count: adjusted_i32(row.transaction_count, transaction_delta_pct, 1),
transaction_price_psm: adjusted_f64(row.transaction_price_psm, price_delta_pct, 1.0),
listing_count: adjusted_i32(row.listing_count, listing_delta_pct, 0),
listing_price_psm: adjusted_f64(row.listing_price_psm, price_delta_pct * 0.85, 1.0),
median_days_on_market: adjusted_f64(
row.median_days_on_market,
-transaction_delta_pct * 0.45 + listing_delta_pct * 0.25,
1.0,
),
rent_price_psm: adjusted_f64(row.rent_price_psm, run.rent_delta_pct * multiplier, 0.1),
new_supply_units: adjusted_i32(row.new_supply_units, run.supply_delta_pct * multiplier, 0),
mortgage_rate_pct: (row.mortgage_rate_pct
+ run.interest_rate_delta_bps * multiplier / 100.0)
.max(0.0),
policy_signal: (row.policy_signal
+ ((run.policy_signal_delta as f64) * multiplier).round() as i32)
.clamp(-2, 2),
..row.clone()
}
}
fn scenario_area_result(
baseline: &ComputedAreaScore,
projected: &ComputedAreaScore,
adjusted: &AreaMetricRow,
run: &ScenarioRun,
multiplier: f64,
) -> ScenarioAreaResult {
ScenarioAreaResult {
area_id: projected.area_id.clone(),
name: projected.name.clone(),
district: projected.district.clone(),
baseline_score: baseline.investment_score,
projected_score: projected.investment_score,
score_delta: round_to(projected.investment_score - baseline.investment_score, 1),
baseline_price_psm: baseline.transaction_price_psm,
projected_price_psm: projected.transaction_price_psm,
price_delta_pct: round_to(
safe_percentage(
projected.transaction_price_psm - baseline.transaction_price_psm,
baseline.transaction_price_psm,
),
1,
),
projected_rent_yield_pct: projected.annual_rent_yield_pct,
projected_liquidity_score: projected.liquidity_score,
projected_supply_risk_score: projected.supply_risk_score,
recommendation: projected.recommendation.clone(),
risk_notes: scenario_risk_notes(projected, adjusted, run, multiplier),
}
}
fn scenario_assumptions(run: &ScenarioRun, multiplier: f64) -> Vec<String> {
vec![
format!("利率变动 {:.0}bp", run.interest_rate_delta_bps * multiplier),
format!(
"政策信号 {}",
((run.policy_signal_delta as f64) * multiplier).round() as i32
),
format!("新增供应 {:.1}%", run.supply_delta_pct * multiplier),
format!("租金变化 {:.1}%", run.rent_delta_pct * multiplier),
format!("价格冲击 {:.1}%", scenario_price_delta_pct(run, multiplier)),
]
}
fn scenario_risk_notes(
projected: &ComputedAreaScore,
adjusted: &AreaMetricRow,
run: &ScenarioRun,
multiplier: f64,
) -> Vec<String> {
let mut notes = Vec::new();
if projected.supply_risk_score >= 70.0 {
notes.push("供应压力偏高,需跟踪新增挂牌与去化速度".to_string());
}
if projected.liquidity_score < 45.0 {
notes.push("流动性承压,议价周期可能拉长".to_string());
}
if projected.annual_rent_yield_pct < 1.7 {
notes.push("租金收益率偏低,安全边际不足".to_string());
}
if adjusted.mortgage_rate_pct >= 4.5 {
notes.push("按揭利率处于偏紧区间,购买力敏感".to_string());
}
if run.policy_signal_delta < 0 && multiplier >= 1.0 {
notes.push("政策假设偏谨慎,估值修复可能延后".to_string());
}
if notes.is_empty() {
notes.push("核心风险未显著恶化,适合继续跟踪量价验证".to_string());
}
notes
}
fn scenario_price_delta_pct(run: &ScenarioRun, multiplier: f64) -> f64 {
let rate_effect = -0.018 * run.interest_rate_delta_bps;
let policy_effect = run.policy_signal_delta as f64 * 1.35;
let supply_effect = -0.035 * run.supply_delta_pct;
let rent_effect = 0.08 * run.rent_delta_pct;
((rate_effect + policy_effect + supply_effect + rent_effect) * multiplier).clamp(-35.0, 35.0)
}
fn scenario_transaction_delta_pct(run: &ScenarioRun, multiplier: f64) -> f64 {
let rate_effect = -0.045 * run.interest_rate_delta_bps;
let policy_effect = run.policy_signal_delta as f64 * 4.5;
let supply_effect = 0.08 * run.supply_delta_pct;
let rent_effect = 0.04 * run.rent_delta_pct;
((rate_effect + policy_effect + supply_effect + rent_effect) * multiplier).clamp(-70.0, 160.0)
}
fn adjusted_f64(value: f64, pct_delta: f64, minimum: f64) -> f64 {
(value * (1.0 + pct_delta / 100.0)).max(minimum)
}
fn adjusted_i32(value: i32, pct_delta: f64, minimum: i32) -> i32 {
((value as f64) * (1.0 + pct_delta / 100.0))
.round()
.max(minimum as f64) as i32
}
async fn recompute_area_scores_in_tx(
tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
month: &str,
@@ -2501,6 +2846,21 @@ fn validate_sha256(value: &str) -> ApiResult<()> {
}
}
fn validate_identifier(value: &str) -> ApiResult<()> {
let valid = !value.is_empty()
&& value.len() <= 80
&& value
.chars()
.all(|item| item.is_ascii_alphanumeric() || item == '_' || item == '-')
&& !has_sensitive_text(value);
if valid {
Ok(())
} else {
Err(ApiError::BadRequest("invalid identifier".to_string()))
}
}
fn has_sensitive_text(value: &str) -> bool {
let lower = value.to_ascii_lowercase();
lower.contains("password")
@@ -2515,11 +2875,15 @@ fn has_sensitive_text(value: &str) -> bool {
#[cfg(test)]
mod tests {
use crate::models::{
CreateDataSource, CreateIngestionRun, CreateRawArtifact, CreateWatchlistItem,
FinishIngestionRun, SimilarNeighborhoodCandidate, UpdateWatchlistItem,
CreateDataSource, CreateIngestionRun, CreateRawArtifact, CreateScenarioRun,
CreateWatchlistItem, FinishIngestionRun, ScenarioRun, SimilarNeighborhoodCandidate,
UpdateWatchlistItem,
};
use super::{parse_compare_ids, similar_neighborhood, validate_month, validate_sha256};
use super::{
parse_compare_ids, scenario_assumptions, scenario_price_delta_pct,
scenario_transaction_delta_pct, similar_neighborhood, validate_month, validate_sha256,
};
#[test]
fn accepts_valid_month() {
@@ -2712,6 +3076,35 @@ mod tests {
.is_err());
}
#[test]
fn validates_scenario_run_payload() {
assert!(CreateScenarioRun {
name: "利率上行压力测试".to_string(),
target_month: "2026-05".to_string(),
area_id: Some("zhangjiang".to_string()),
interest_rate_delta_bps: 50.0,
policy_signal_delta: -1,
supply_delta_pct: 20.0,
rent_delta_pct: 3.0,
notes: Some("手工假设".to_string()),
}
.validate()
.is_ok());
assert!(CreateScenarioRun {
name: "bad".to_string(),
target_month: "2026-05".to_string(),
area_id: None,
interest_rate_delta_bps: 500.0,
policy_signal_delta: 0,
supply_delta_pct: 0.0,
rent_delta_pct: 0.0,
notes: None,
}
.validate()
.is_err());
}
#[test]
fn validates_sha256_query_value() {
assert!(validate_sha256(&"0".repeat(64)).is_ok());
@@ -2779,4 +3172,24 @@ mod tests {
building_age_score: 85.0,
}
}
#[test]
fn scenario_shocks_have_expected_direction() {
let run = ScenarioRun {
scenario_run_id: 1,
name: "压力测试".to_string(),
target_month: "2026-05".to_string(),
area_id: None,
interest_rate_delta_bps: 50.0,
policy_signal_delta: -1,
supply_delta_pct: 30.0,
rent_delta_pct: 5.0,
notes: "".to_string(),
created_at: "2026-06-24".to_string(),
};
assert!(scenario_price_delta_pct(&run, 1.0) < 0.0);
assert!(scenario_transaction_delta_pct(&run, 1.0) < 0.0);
assert_eq!(scenario_assumptions(&run, 1.0).len(), 5);
}
}

View File

@@ -59,6 +59,12 @@ pub struct ModelRunDiffQuery {
pub candidate_run_id: i64,
}
#[derive(Debug, Deserialize)]
pub struct ScenarioRunQuery {
pub target_month: Option<String>,
pub area_id: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct RawArtifactQuery {
pub source_id: Option<i64>,
@@ -487,6 +493,100 @@ pub struct SimilarNeighborhoodResponse {
pub items: Vec<SimilarNeighborhood>,
}
#[derive(Debug, Clone, Serialize, FromRow)]
pub struct ScenarioRun {
pub scenario_run_id: i64,
pub name: String,
pub target_month: String,
pub area_id: Option<String>,
pub interest_rate_delta_bps: f64,
pub policy_signal_delta: i32,
pub supply_delta_pct: f64,
pub rent_delta_pct: f64,
pub notes: String,
pub created_at: String,
}
#[derive(Debug, Deserialize)]
pub struct CreateScenarioRun {
pub name: String,
pub target_month: String,
pub area_id: Option<String>,
pub interest_rate_delta_bps: f64,
pub policy_signal_delta: i32,
pub supply_delta_pct: f64,
pub rent_delta_pct: f64,
pub notes: Option<String>,
}
impl CreateScenarioRun {
pub fn validate(&self) -> Result<(), &'static str> {
if self.name.trim().is_empty() {
return Err("name is required");
}
if self.name.len() > 80 {
return Err("name is too long");
}
if !(-300.0..=300.0).contains(&self.interest_rate_delta_bps) {
return Err("interest_rate_delta_bps must be between -300 and 300");
}
if !(-4..=4).contains(&self.policy_signal_delta) {
return Err("policy_signal_delta must be between -4 and 4");
}
if !(-100.0..=500.0).contains(&self.supply_delta_pct) {
return Err("supply_delta_pct must be between -100 and 500");
}
if !(-80.0..=200.0).contains(&self.rent_delta_pct) {
return Err("rent_delta_pct must be between -80 and 200");
}
for value in [
Some(self.name.as_str()),
self.area_id.as_deref(),
self.notes.as_deref(),
]
.into_iter()
.flatten()
{
if has_sensitive_text(value) {
return Err("payload contains sensitive text");
}
}
Ok(())
}
}
#[derive(Debug, Serialize)]
pub struct ScenarioVariant {
pub key: String,
pub name: String,
pub assumptions: Vec<String>,
pub items: Vec<ScenarioAreaResult>,
}
#[derive(Debug, Serialize)]
pub struct ScenarioAreaResult {
pub area_id: String,
pub name: String,
pub district: String,
pub baseline_score: f64,
pub projected_score: f64,
pub score_delta: f64,
pub baseline_price_psm: f64,
pub projected_price_psm: f64,
pub price_delta_pct: f64,
pub projected_rent_yield_pct: f64,
pub projected_liquidity_score: f64,
pub projected_supply_risk_score: f64,
pub recommendation: String,
pub risk_notes: Vec<String>,
}
#[derive(Debug, Serialize)]
pub struct ScenarioRunResponse {
pub run: ScenarioRun,
pub variants: Vec<ScenarioVariant>,
}
#[derive(Debug, Clone, Serialize, FromRow)]
pub struct DataSource {
pub source_id: i64,

View File

@@ -5,13 +5,13 @@ use tower_http::trace::TraceLayer;
use crate::handlers::{
archive_watchlist_item, compare_areas, compare_neighborhoods, create_area_score_model_run,
create_data_source, create_ingestion_run, create_raw_artifact, create_watchlist_event,
create_watchlist_item, diff_area_score_model_runs, execute_import, finish_ingestion_run,
get_area_detail, get_area_diagnostics, get_area_score_lineage, get_neighborhood,
get_neighborhood_detail, get_similar_neighborhoods, health, list_area_scores,
create_data_source, create_ingestion_run, create_raw_artifact, create_scenario_run,
create_watchlist_event, create_watchlist_item, diff_area_score_model_runs, execute_import,
finish_ingestion_run, get_area_detail, get_area_diagnostics, get_area_score_lineage,
get_neighborhood, get_neighborhood_detail, get_similar_neighborhoods, health, list_area_scores,
list_data_sources, list_ingestion_runs, list_model_runs, list_neighborhood_scores,
list_neighborhoods, list_raw_artifacts, list_watchlist_events, list_watchlist_items,
market_overview, ready, update_watchlist_item, validate_import,
list_neighborhoods, list_raw_artifacts, list_scenario_runs, list_watchlist_events,
list_watchlist_items, market_overview, ready, update_watchlist_item, validate_import,
};
use crate::state::AppState;
@@ -81,6 +81,10 @@ pub fn build_router(state: AppState) -> Router {
.route(
"/watchlist/{watchlist_item_id}/events",
get(list_watchlist_events).post(create_watchlist_event),
)
.route(
"/scenario-runs",
get(list_scenario_runs).post(create_scenario_run),
),
)
.layer(TraceLayer::new_for_http())