huheng-research/apps/api/src/handlers.rs

use axum::extract::{Path, Query, State};
use axum::Json;
use serde::Serialize;

use crate::error::{ApiError, ApiResult};
use crate::import_templates::{execute_import_payload, validate_import_payload};
use crate::models::{
    AreaComparison, AreaDetail, AreaDiagnostics, AreaDiagnosticsQuery, AreaMonthlyMetric,
    AreaProfile, AreaScore, AreaScoreLineage, AreaScoreLineageQuery, AreaScoreModelRunResponse,
    AreaScoreRunDiff, AreaScoreRunDiffItem, AreaScoreSummary, CompareQuery, ComparisonMetric,
    ComparisonMetricValue, CreateAreaScoreModelRun, CreateDataSource, CreateIngestionRun,
    CreateRawArtifact, CreateReport, CreateReportGenerationRun, 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, Report,
    ReportGenerationRun, ReportGenerationRunQuery, ReportGenerationRunResponse, ReportQuery,
    ReportTemplate, 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;

#[derive(Debug, Serialize)]
pub struct HealthResponse {
    pub status: &'static str,
    pub service: &'static str,
}

pub async fn health() -> Json<HealthResponse> {
    Json(HealthResponse {
        status: "ok",
        service: "shanghai-housing-api",
    })
}

pub async fn ready(State(state): State<AppState>) -> ApiResult<Json<HealthResponse>> {
    sqlx::query("select 1").execute(&state.pool).await?;
    Ok(Json(HealthResponse {
        status: "ok",
        service: "shanghai-housing-api",
    }))
}

pub async fn validate_import(
    Json(payload): Json<ImportValidationRequest>,
) -> ApiResult<Json<ImportValidationResponse>> {
    let result = validate_import_payload(&payload).map_err(ApiError::BadRequest)?;
    Ok(Json(result))
}

pub async fn execute_import(
    State(state): State<AppState>,
    Json(payload): Json<ImportExecutionRequest>,
) -> ApiResult<Json<ImportExecutionResponse>> {
    let mut tx = state.pool.begin().await?;
    let result = execute_import_payload(&mut tx, &payload).await;

    match result {
        Ok(response) => {
            sqlx::query(
                r#"
                UPDATE audit.ingestion_runs
                SET status = 'succeeded',
                    finished_at = now(),
                    row_count = $2,
                    raw_artifact_id = $3,
                    error_message = NULL
                WHERE run_id = $1
                "#,
            )
            .bind(payload.run_id)
            .bind(response.row_count as i32)
            .bind(payload.raw_artifact_id)
            .execute(&mut *tx)
            .await?;
            tx.commit().await?;
            Ok(Json(response))
        }
        Err(message) => {
            tx.rollback().await?;
            sqlx::query(
                r#"
                UPDATE audit.ingestion_runs
                SET status = 'failed',
                    finished_at = now(),
                    error_message = $2
                WHERE run_id = $1
                "#,
            )
            .bind(payload.run_id)
            .bind(message.clone())
            .execute(&state.pool)
            .await?;
            Err(ApiError::BadRequest(message))
        }
    }
}

pub async fn list_area_scores(
    State(state): State<AppState>,
    Query(query): Query<MonthQuery>,
) -> ApiResult<Json<Vec<AreaScore>>> {
    validate_month(&query.month)?;
    let scores = fetch_area_scores(&state, &query.month).await?;
    Ok(Json(scores))
}

pub async fn market_overview(
    State(state): State<AppState>,
    Query(query): Query<MonthQuery>,
) -> ApiResult<Json<MarketOverview>> {
    validate_month(&query.month)?;
    let scores = fetch_area_scores(&state, &query.month).await?;
    let area_count = scores.len();
    let average_investment_score = average(scores.iter().map(|score| score.investment_score));
    let average_rent_yield_pct = average(scores.iter().map(|score| score.annual_rent_yield_pct));
    let top_area = scores.first().map(AreaScoreSummary::from);
    let weakest_area = scores.last().map(AreaScoreSummary::from);
    let highest_supply_risk_area = scores
        .iter()
        .max_by(|left, right| {
            left.supply_risk_score
                .partial_cmp(&right.supply_risk_score)
                .unwrap_or(std::cmp::Ordering::Equal)
        })
        .map(AreaScoreSummary::from);

    Ok(Json(MarketOverview {
        month: query.month,
        area_count,
        average_investment_score,
        average_rent_yield_pct,
        top_area,
        weakest_area,
        highest_supply_risk_area,
    }))
}

pub async fn compare_areas(
    State(state): State<AppState>,
    Query(query): Query<CompareQuery>,
) -> ApiResult<Json<AreaComparison>> {
    validate_month(&query.month)?;
    let requested_ids = parse_compare_ids(&query.ids)?;
    let scores = sqlx::query_as::<_, AreaScore>(
        r#"
        SELECT
            area_id,
            name,
            district,
            segment,
            month,
            investment_score,
            recommendation,
            liquidity_score,
            momentum_score,
            rent_support_score,
            safety_margin_score,
            credit_support_score,
            supply_risk_score,
            valuation_pressure_score,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            rent_price_psm,
            median_days_on_market,
            annual_rent_yield_pct,
            listing_pressure_ratio,
            discount_pct,
            price_momentum_pct,
            volume_momentum_pct
        FROM gold.area_scores
        WHERE month = $1
          AND area_id = ANY($2)
        "#,
    )
    .bind(&query.month)
    .bind(&requested_ids)
    .fetch_all(&state.pool)
    .await?;

    let items = sort_by_requested_order(scores, &requested_ids, |score| &score.area_id);
    let missing_ids = missing_ids(&requested_ids, &items, |score| &score.area_id);
    let metrics = area_comparison_metrics(&items);

    Ok(Json(AreaComparison {
        month: query.month,
        requested_ids,
        missing_ids,
        items,
        metrics,
    }))
}

pub async fn compare_neighborhoods(
    State(state): State<AppState>,
    Query(query): Query<CompareQuery>,
) -> ApiResult<Json<NeighborhoodComparison>> {
    validate_month(&query.month)?;
    let requested_ids = parse_compare_ids(&query.ids)?;
    let scores = sqlx::query_as::<_, NeighborhoodComparisonItem>(
        r#"
        SELECT
            s.neighborhood_id,
            s.area_id,
            s.name,
            s.area_name,
            s.district,
            n.built_year,
            n.property_type,
            n.metro_distance_m,
            n.school_quality,
            s.month,
            s.investment_score,
            s.recommendation,
            s.liquidity_score,
            s.rent_support_score,
            s.price_safety_score,
            s.location_score,
            s.building_age_score,
            s.transaction_count,
            s.transaction_price_psm,
            s.listing_count,
            s.listing_price_psm,
            s.rent_price_psm,
            s.median_days_on_market,
            s.annual_rent_yield_pct,
            s.discount_pct,
            s.available_units
        FROM gold.neighborhood_scores s
        JOIN silver.neighborhoods n ON n.neighborhood_id = s.neighborhood_id
        WHERE s.month = $1
          AND s.neighborhood_id = ANY($2)
        "#,
    )
    .bind(&query.month)
    .bind(&requested_ids)
    .fetch_all(&state.pool)
    .await?;

    let items = sort_by_requested_order(scores, &requested_ids, |score| &score.neighborhood_id);
    let missing_ids = missing_ids(&requested_ids, &items, |score| &score.neighborhood_id);
    let metrics = neighborhood_comparison_metrics(&items);

    Ok(Json(NeighborhoodComparison {
        month: query.month,
        requested_ids,
        missing_ids,
        items,
        metrics,
    }))
}

pub async fn list_model_runs(
    State(state): State<AppState>,
    Query(query): Query<ModelRunQuery>,
) -> ApiResult<Json<Vec<ModelRun>>> {
    if let Some(month) = query.target_month.as_deref() {
        validate_month(month)?;
    }
    if let Some(model_name) = query.model_name.as_deref() {
        if has_sensitive_text(model_name) {
            return Err(ApiError::BadRequest(
                "model_name contains sensitive text".to_string(),
            ));
        }
    }

    let runs = sqlx::query_as::<_, ModelRun>(
        r#"
        SELECT
            model_run_id,
            model_name,
            model_version,
            target_month,
            status,
            parameters,
            started_at::text AS started_at,
            finished_at::text AS finished_at,
            row_count,
            error_message
        FROM gold.model_runs
        WHERE ($1::text IS NULL OR model_name = $1)
          AND ($2::text IS NULL OR target_month = $2)
        ORDER BY started_at DESC, model_run_id DESC
        LIMIT 100
        "#,
    )
    .bind(query.model_name)
    .bind(query.target_month)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(runs))
}

pub async fn diff_area_score_model_runs(
    State(state): State<AppState>,
    Query(query): Query<ModelRunDiffQuery>,
) -> ApiResult<Json<AreaScoreRunDiff>> {
    if query.base_run_id == query.candidate_run_id {
        return Err(ApiError::BadRequest(
            "base_run_id and candidate_run_id must be different".to_string(),
        ));
    }

    let base_run = fetch_model_run(&state, query.base_run_id)
        .await?
        .ok_or_else(|| ApiError::BadRequest("base model run not found".to_string()))?;
    let candidate_run = fetch_model_run(&state, query.candidate_run_id)
        .await?
        .ok_or_else(|| ApiError::BadRequest("candidate model run not found".to_string()))?;
    if base_run.model_name != "area_scores" || candidate_run.model_name != "area_scores" {
        return Err(ApiError::BadRequest(
            "both model runs must be area_scores runs".to_string(),
        ));
    }

    let items = sqlx::query_as::<_, AreaScoreRunDiffItem>(
        r#"
        SELECT
            COALESCE(base.area_id, candidate.area_id) AS area_id,
            COALESCE(candidate.name, base.name) AS name,
            COALESCE(candidate.district, base.district) AS district,
            base.investment_score AS base_score,
            candidate.investment_score AS candidate_score,
            CASE
                WHEN base.investment_score IS NULL OR candidate.investment_score IS NULL THEN NULL
                ELSE candidate.investment_score - base.investment_score
            END AS score_delta,
            base.recommendation AS base_recommendation,
            candidate.recommendation AS candidate_recommendation
        FROM (
            SELECT area_id, name, district, investment_score, recommendation
            FROM gold.area_score_snapshots
            WHERE model_run_id = $1
        ) base
        FULL JOIN (
            SELECT area_id, name, district, investment_score, recommendation
            FROM gold.area_score_snapshots
            WHERE model_run_id = $2
        ) candidate ON candidate.area_id = base.area_id
        ORDER BY ABS(
            COALESCE(candidate.investment_score, 0) - COALESCE(base.investment_score, 0)
        ) DESC,
        area_id ASC
        "#,
    )
    .bind(query.base_run_id)
    .bind(query.candidate_run_id)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(AreaScoreRunDiff {
        base_run,
        candidate_run,
        items,
    }))
}

pub async fn create_area_score_model_run(
    State(state): State<AppState>,
    Json(payload): Json<CreateAreaScoreModelRun>,
) -> ApiResult<Json<AreaScoreModelRunResponse>> {
    validate_month(&payload.month)?;
    let model_version = payload
        .model_version
        .clone()
        .unwrap_or_else(|| "area-score-rust-v1".to_string());
    if has_sensitive_text(&model_version) {
        return Err(ApiError::BadRequest(
            "model_version contains sensitive text".to_string(),
        ));
    }

    let mut tx = state.pool.begin().await?;
    let model_run = sqlx::query_as::<_, ModelRun>(
        r#"
        INSERT INTO gold.model_runs (
            model_name,
            model_version,
            target_month,
            status,
            parameters
        )
        VALUES ('area_scores', $1, $2, 'running', $3)
        RETURNING
            model_run_id,
            model_name,
            model_version,
            target_month,
            status,
            parameters,
            started_at::text AS started_at,
            finished_at::text AS finished_at,
            row_count,
            error_message
        "#,
    )
    .bind(&model_version)
    .bind(&payload.month)
    .bind(serde_json::json!({
        "source_table": "silver.area_monthly_metrics",
        "weights": {
            "liquidity": 0.25,
            "momentum": 0.20,
            "rent_support": 0.20,
            "safety_margin": 0.15,
            "credit_support": 0.10,
            "supply_risk_inverse": 0.10
        }
    }))
    .fetch_one(&mut *tx)
    .await?;

    let result = recompute_area_scores_in_tx(
        &mut tx,
        &payload.month,
        &model_version,
        model_run.model_run_id,
    )
    .await;

    match result {
        Ok(row_count) => {
            let completed_run = finish_model_run(
                &mut tx,
                model_run.model_run_id,
                "succeeded",
                Some(row_count),
                None,
            )
            .await?;
            tx.commit().await?;
            let scores = fetch_area_scores(&state, &payload.month).await?;
            Ok(Json(AreaScoreModelRunResponse {
                model_run: completed_run,
                scores,
            }))
        }
        Err(message) => {
            let _ = finish_model_run(
                &mut tx,
                model_run.model_run_id,
                "failed",
                None,
                Some(&message),
            )
            .await;
            tx.commit().await?;
            Err(ApiError::BadRequest(message))
        }
    }
}

pub async fn get_area_score_lineage(
    State(state): State<AppState>,
    Path(area_id): Path<String>,
    Query(query): Query<AreaScoreLineageQuery>,
) -> ApiResult<Json<AreaScoreLineage>> {
    validate_month(&query.month)?;
    let lineage = fetch_area_score_lineage(&state, &area_id, &query.month)
        .await?
        .ok_or_else(|| ApiError::BadRequest("area score lineage not found".to_string()))?;

    Ok(Json(lineage))
}

async fn fetch_area_score_lineage(
    state: &AppState,
    area_id: &str,
    month: &str,
) -> Result<Option<AreaScoreLineage>, sqlx::Error> {
    sqlx::query_as::<_, AreaScoreLineage>(
        r#"
        SELECT
            s.area_id,
            a.name AS area_name,
            s.month,
            g.investment_score,
            g.model_run_id,
            g.model_version,
            s.ingestion_run_id,
            s.raw_artifact_id,
            ra.raw_uri,
            ra.sha256,
            ds.name AS source_name,
            s.updated_at::text AS metric_updated_at,
            g.computed_at::text AS score_computed_at
        FROM gold.area_scores g
        JOIN silver.area_monthly_metrics s
          ON s.area_id = g.area_id
         AND s.month = g.month
        JOIN silver.areas a ON a.area_id = s.area_id
        LEFT JOIN audit.raw_artifacts ra ON ra.artifact_id = s.raw_artifact_id
        LEFT JOIN audit.data_sources ds ON ds.source_id = ra.source_id
        WHERE s.area_id = $1
          AND s.month = $2
        "#,
    )
    .bind(area_id)
    .bind(month)
    .fetch_optional(&state.pool)
    .await
}

pub async fn get_area_detail(
    State(state): State<AppState>,
    Path(area_id): Path<String>,
    Query(query): Query<MonthQuery>,
) -> ApiResult<Json<AreaDetail>> {
    validate_month(&query.month)?;

    let profile = sqlx::query_as::<_, AreaProfile>(
        r#"
        SELECT area_id, name, district, segment, notes
        FROM silver.areas
        WHERE area_id = $1
        "#,
    )
    .bind(&area_id)
    .fetch_optional(&state.pool)
    .await?
    .ok_or_else(|| ApiError::BadRequest("area not found".to_string()))?;

    let current_score = sqlx::query_as::<_, AreaScore>(
        r#"
        SELECT
            area_id,
            name,
            district,
            segment,
            month,
            investment_score,
            recommendation,
            liquidity_score,
            momentum_score,
            rent_support_score,
            safety_margin_score,
            credit_support_score,
            supply_risk_score,
            valuation_pressure_score,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            rent_price_psm,
            median_days_on_market,
            annual_rent_yield_pct,
            listing_pressure_ratio,
            discount_pct,
            price_momentum_pct,
            volume_momentum_pct
        FROM gold.area_scores
        WHERE area_id = $1
          AND month = $2
        "#,
    )
    .bind(&area_id)
    .bind(&query.month)
    .fetch_optional(&state.pool)
    .await?;

    let score_lineage = fetch_area_score_lineage(&state, &area_id, &query.month).await?;
    let monthly_metrics = sqlx::query_as::<_, AreaMonthlyMetric>(
        r#"
        SELECT
            area_id,
            month,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            median_days_on_market,
            rent_price_psm,
            new_supply_units,
            land_residential_gfa_sqm,
            mortgage_rate_pct,
            policy_signal,
            source,
            ingestion_run_id,
            raw_artifact_id
        FROM silver.area_monthly_metrics
        WHERE area_id = $1
          AND month <= $2
        ORDER BY month DESC
        LIMIT 12
        "#,
    )
    .bind(&area_id)
    .bind(&query.month)
    .fetch_all(&state.pool)
    .await?;

    let diagnostics = fetch_area_diagnostics(&state, &area_id, &query.month).await?;
    let neighborhood_scores =
        fetch_neighborhood_scores_for_area(&state, &area_id, &query.month).await?;

    Ok(Json(AreaDetail {
        profile,
        current_score,
        score_lineage,
        monthly_metrics,
        diagnostics,
        neighborhood_scores,
    }))
}

pub async fn get_area_diagnostics(
    State(state): State<AppState>,
    Path(area_id): Path<String>,
    Query(query): Query<AreaDiagnosticsQuery>,
) -> ApiResult<Json<AreaDiagnostics>> {
    validate_month(&query.month)?;
    let diagnostics = fetch_area_diagnostics(&state, &area_id, &query.month)
        .await?
        .ok_or_else(|| ApiError::BadRequest("area diagnostics not found".to_string()))?;

    Ok(Json(diagnostics))
}

async fn fetch_area_diagnostics(
    state: &AppState,
    area_id: &str,
    month: &str,
) -> Result<Option<AreaDiagnostics>, sqlx::Error> {
    let rows = sqlx::query_as::<_, AreaMonthlyMetric>(
        r#"
        SELECT
            area_id,
            month,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            median_days_on_market,
            rent_price_psm,
            new_supply_units,
            land_residential_gfa_sqm,
            mortgage_rate_pct,
            policy_signal,
            source,
            ingestion_run_id,
            raw_artifact_id
        FROM silver.area_monthly_metrics
        WHERE area_id = $1
          AND month <= $2
        ORDER BY month ASC
        "#,
    )
    .bind(area_id)
    .bind(month)
    .fetch_all(&state.pool)
    .await?;

    let Some(current) = rows.iter().find(|row| row.month == month) else {
        return Ok(None);
    };

    let transaction_price_values = rows
        .iter()
        .map(|row| row.transaction_price_psm)
        .collect::<Vec<_>>();
    let transaction_count_values = rows
        .iter()
        .map(|row| row.transaction_count as f64)
        .collect::<Vec<_>>();
    let rent_yield_values = rows
        .iter()
        .map(|row| safe_percentage(row.rent_price_psm * 12.0, row.transaction_price_psm))
        .collect::<Vec<_>>();
    let inventory_pressure_values = rows
        .iter()
        .map(|row| safe_ratio(row.listing_count as f64, row.transaction_count as f64))
        .collect::<Vec<_>>();

    let metrics = vec![
        diagnostic_metric(
            "transaction_price_psm",
            "成交均价",
            "元/㎡",
            current.transaction_price_psm,
            &transaction_price_values,
        ),
        diagnostic_metric(
            "transaction_count",
            "成交量",
            "套",
            current.transaction_count as f64,
            &transaction_count_values,
        ),
        diagnostic_metric(
            "annual_rent_yield_pct",
            "租金收益率",
            "%",
            safe_percentage(current.rent_price_psm * 12.0, current.transaction_price_psm),
            &rent_yield_values,
        ),
        diagnostic_metric(
            "inventory_pressure_ratio",
            "库存压力",
            "倍",
            safe_ratio(
                current.listing_count as f64,
                current.transaction_count as f64,
            ),
            &inventory_pressure_values,
        ),
    ];
    let anomaly_count = metrics.iter().filter(|metric| metric.anomaly).count();

    Ok(Some(AreaDiagnostics {
        area_id: area_id.to_string(),
        month: month.to_string(),
        metrics,
        anomaly_count,
    }))
}

async fn fetch_neighborhood_scores_for_area(
    state: &AppState,
    area_id: &str,
    month: &str,
) -> Result<Vec<NeighborhoodScore>, sqlx::Error> {
    sqlx::query_as::<_, NeighborhoodScore>(
        r#"
        SELECT
            neighborhood_id,
            area_id,
            name,
            area_name,
            district,
            month,
            investment_score,
            recommendation,
            liquidity_score,
            rent_support_score,
            price_safety_score,
            location_score,
            building_age_score,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            rent_price_psm,
            median_days_on_market,
            annual_rent_yield_pct,
            discount_pct,
            available_units
        FROM gold.neighborhood_scores
        WHERE area_id = $1
          AND month = $2
        ORDER BY investment_score DESC, neighborhood_id ASC
        "#,
    )
    .bind(area_id)
    .bind(month)
    .fetch_all(&state.pool)
    .await
}

pub async fn list_neighborhoods(
    State(state): State<AppState>,
    Query(query): Query<NeighborhoodQuery>,
) -> ApiResult<Json<Vec<Neighborhood>>> {
    let neighborhoods = sqlx::query_as::<_, Neighborhood>(
        r#"
        SELECT
            n.neighborhood_id,
            n.area_id,
            a.name AS area_name,
            a.district,
            n.name,
            n.built_year,
            n.property_type,
            n.metro_distance_m,
            n.school_quality,
            n.notes
        FROM silver.neighborhoods n
        JOIN silver.areas a ON a.area_id = n.area_id
        WHERE ($1::text IS NULL OR n.area_id = $1)
        ORDER BY a.district, a.name, n.name
        "#,
    )
    .bind(query.area_id)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(neighborhoods))
}

pub async fn get_neighborhood(
    State(state): State<AppState>,
    Path(neighborhood_id): Path<String>,
) -> ApiResult<Json<Neighborhood>> {
    let neighborhood = fetch_neighborhood_profile(&state, &neighborhood_id)
        .await?
        .ok_or_else(|| ApiError::BadRequest("neighborhood not found".to_string()))?;

    Ok(Json(neighborhood))
}

pub async fn get_neighborhood_detail(
    State(state): State<AppState>,
    Path(neighborhood_id): Path<String>,
    Query(query): Query<MonthQuery>,
) -> ApiResult<Json<NeighborhoodDetail>> {
    validate_month(&query.month)?;

    let profile = fetch_neighborhood_profile(&state, &neighborhood_id)
        .await?
        .ok_or_else(|| ApiError::BadRequest("neighborhood not found".to_string()))?;

    let current_score = sqlx::query_as::<_, NeighborhoodScore>(
        r#"
        SELECT
            neighborhood_id,
            area_id,
            name,
            area_name,
            district,
            month,
            investment_score,
            recommendation,
            liquidity_score,
            rent_support_score,
            price_safety_score,
            location_score,
            building_age_score,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            rent_price_psm,
            median_days_on_market,
            annual_rent_yield_pct,
            discount_pct,
            available_units
        FROM gold.neighborhood_scores
        WHERE neighborhood_id = $1
          AND month = $2
        "#,
    )
    .bind(&neighborhood_id)
    .bind(&query.month)
    .fetch_optional(&state.pool)
    .await?;

    let monthly_metrics = sqlx::query_as::<_, NeighborhoodMonthlyMetric>(
        r#"
        SELECT
            neighborhood_id,
            month,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            rent_price_psm,
            median_days_on_market,
            available_units,
            source,
            ingestion_run_id,
            raw_artifact_id
        FROM silver.neighborhood_monthly_metrics
        WHERE neighborhood_id = $1
          AND month <= $2
        ORDER BY month DESC
        LIMIT 12
        "#,
    )
    .bind(&neighborhood_id)
    .bind(&query.month)
    .fetch_all(&state.pool)
    .await?;

    let watchlist_items = sqlx::query_as::<_, WatchlistItem>(
        r#"
        SELECT
            watchlist_item_id,
            neighborhood_id,
            area_id,
            target_price_psm,
            status,
            label,
            priority,
            trigger_operator,
            trigger_price_psm,
            notes,
            created_at::text AS created_at,
            updated_at::text AS updated_at,
            last_reviewed_at::text AS last_reviewed_at
        FROM app.watchlist_items
        WHERE neighborhood_id = $1
          AND status <> 'archived'
        ORDER BY updated_at DESC, watchlist_item_id DESC
        "#,
    )
    .bind(&neighborhood_id)
    .fetch_all(&state.pool)
    .await?;
    let similar_neighborhoods =
        fetch_similar_neighborhoods(&state, &neighborhood_id, &query.month, 6)
            .await?
            .map(|response| response.items)
            .unwrap_or_default();

    Ok(Json(NeighborhoodDetail {
        profile,
        current_score,
        monthly_metrics,
        similar_neighborhoods,
        watchlist_items,
    }))
}

pub async fn get_similar_neighborhoods(
    State(state): State<AppState>,
    Path(neighborhood_id): Path<String>,
    Query(query): Query<SimilarNeighborhoodQuery>,
) -> ApiResult<Json<SimilarNeighborhoodResponse>> {
    validate_month(&query.month)?;
    let limit = query.limit.unwrap_or(8).clamp(1, 10);
    let response = fetch_similar_neighborhoods(&state, &neighborhood_id, &query.month, limit)
        .await?
        .ok_or_else(|| ApiError::BadRequest("neighborhood score not found".to_string()))?;

    Ok(Json(response))
}

async fn fetch_similar_neighborhoods(
    state: &AppState,
    neighborhood_id: &str,
    month: &str,
    limit: i64,
) -> Result<Option<SimilarNeighborhoodResponse>, sqlx::Error> {
    let candidates = sqlx::query_as::<_, SimilarNeighborhoodCandidate>(
        r#"
        SELECT
            s.neighborhood_id,
            s.area_id,
            s.name,
            s.area_name,
            s.district,
            n.built_year,
            n.property_type,
            n.metro_distance_m,
            n.school_quality,
            s.month,
            s.investment_score,
            s.recommendation,
            s.transaction_price_psm,
            s.annual_rent_yield_pct,
            s.liquidity_score,
            s.location_score,
            s.building_age_score
        FROM gold.neighborhood_scores s
        JOIN silver.neighborhoods n ON n.neighborhood_id = s.neighborhood_id
        WHERE s.month = $1
        "#,
    )
    .bind(month)
    .fetch_all(&state.pool)
    .await?;

    let Some(target) = candidates
        .iter()
        .find(|candidate| candidate.neighborhood_id == neighborhood_id)
        .cloned()
    else {
        return Ok(None);
    };

    let mut items = candidates
        .into_iter()
        .filter(|candidate| candidate.neighborhood_id != neighborhood_id)
        .map(|candidate| similar_neighborhood(&target, candidate))
        .collect::<Vec<_>>();
    items.sort_by(|left, right| {
        right
            .similarity_score
            .partial_cmp(&left.similarity_score)
            .unwrap_or(std::cmp::Ordering::Equal)
    });
    items.truncate(limit as usize);

    Ok(Some(SimilarNeighborhoodResponse {
        target,
        month: month.to_string(),
        items,
    }))
}

fn similar_neighborhood(
    target: &SimilarNeighborhoodCandidate,
    candidate: SimilarNeighborhoodCandidate,
) -> SimilarNeighborhood {
    let mut similarity_score = 0.0;
    let mut reasons = Vec::new();

    if candidate.property_type == target.property_type {
        similarity_score += 22.0;
        reasons.push("property_type_match".to_string());
    }

    if candidate.area_id == target.area_id {
        similarity_score += 20.0;
        reasons.push("same_area".to_string());
    } else if candidate.district == target.district {
        similarity_score += 10.0;
        reasons.push("same_district".to_string());
    }

    let price_gap_pct = safe_percentage(
        (candidate.transaction_price_psm - target.transaction_price_psm).abs(),
        target.transaction_price_psm,
    );
    similarity_score += closeness_score(price_gap_pct, 30.0) * 24.0;
    reasons.push(format!("price_gap_pct:{:.1}", round_to(price_gap_pct, 1)));

    if let (Some(candidate_year), Some(target_year)) = (candidate.built_year, target.built_year) {
        let age_gap = (candidate_year - target_year).abs() as f64;
        similarity_score += closeness_score(age_gap, 20.0) * 18.0;
        reasons.push(format!("building_age_gap_years:{}", age_gap.round() as i32));
    }

    if let (Some(candidate_distance), Some(target_distance)) =
        (candidate.metro_distance_m, target.metro_distance_m)
    {
        let metro_gap = (candidate_distance - target_distance).abs() as f64;
        similarity_score += closeness_score(metro_gap, 1500.0) * 16.0;
        reasons.push(format!("metro_gap_m:{}", metro_gap.round() as i32));
    }

    SimilarNeighborhood {
        neighborhood_id: candidate.neighborhood_id,
        area_id: candidate.area_id,
        name: candidate.name,
        area_name: candidate.area_name,
        district: candidate.district,
        built_year: candidate.built_year,
        property_type: candidate.property_type,
        metro_distance_m: candidate.metro_distance_m,
        school_quality: candidate.school_quality,
        investment_score: candidate.investment_score,
        recommendation: candidate.recommendation,
        transaction_price_psm: candidate.transaction_price_psm,
        annual_rent_yield_pct: candidate.annual_rent_yield_pct,
        similarity_score: round_to(similarity_score.min(100.0), 1),
        relative_price_pct: round_to(
            safe_percentage(
                candidate.transaction_price_psm - target.transaction_price_psm,
                target.transaction_price_psm,
            ),
            1,
        ),
        reasons,
    }
}

fn closeness_score(gap: f64, max_gap: f64) -> f64 {
    if max_gap <= 0.0 {
        return 0.0;
    }
    (1.0 - gap / max_gap).clamp(0.0, 1.0)
}

async fn fetch_neighborhood_profile(
    state: &AppState,
    neighborhood_id: &str,
) -> Result<Option<Neighborhood>, sqlx::Error> {
    sqlx::query_as::<_, Neighborhood>(
        r#"
        SELECT
            n.neighborhood_id,
            n.area_id,
            a.name AS area_name,
            a.district,
            n.name,
            n.built_year,
            n.property_type,
            n.metro_distance_m,
            n.school_quality,
            n.notes
        FROM silver.neighborhoods n
        JOIN silver.areas a ON a.area_id = n.area_id
        WHERE n.neighborhood_id = $1
        "#,
    )
    .bind(neighborhood_id)
    .fetch_optional(&state.pool)
    .await
}

pub async fn list_neighborhood_scores(
    State(state): State<AppState>,
    Query(query): Query<NeighborhoodScoreQuery>,
) -> ApiResult<Json<Vec<NeighborhoodScore>>> {
    validate_month(&query.month)?;

    let scores = sqlx::query_as::<_, NeighborhoodScore>(
        r#"
        SELECT
            neighborhood_id,
            area_id,
            name,
            area_name,
            district,
            month,
            investment_score,
            recommendation,
            liquidity_score,
            rent_support_score,
            price_safety_score,
            location_score,
            building_age_score,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            rent_price_psm,
            median_days_on_market,
            annual_rent_yield_pct,
            discount_pct,
            available_units
        FROM gold.neighborhood_scores
        WHERE month = $1
          AND ($2::text IS NULL OR area_id = $2)
        ORDER BY investment_score DESC, neighborhood_id ASC
        "#,
    )
    .bind(query.month)
    .bind(query.area_id)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(scores))
}

pub async fn list_data_sources(State(state): State<AppState>) -> ApiResult<Json<Vec<DataSource>>> {
    let sources = sqlx::query_as::<_, DataSource>(
        r#"
        SELECT
            source_id,
            name,
            source_type,
            url,
            cadence,
            reliability,
            notes,
            created_at::text AS created_at
        FROM audit.data_sources
        ORDER BY name ASC
        "#,
    )
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(sources))
}

pub async fn create_data_source(
    State(state): State<AppState>,
    Json(payload): Json<CreateDataSource>,
) -> ApiResult<Json<DataSource>> {
    payload
        .validate()
        .map_err(|message| ApiError::BadRequest(message.to_string()))?;

    let source = sqlx::query_as::<_, DataSource>(
        r#"
        INSERT INTO audit.data_sources (
            name,
            source_type,
            url,
            cadence,
            reliability,
            notes
        )
        VALUES ($1, $2, $3, $4, COALESCE($5, 'unknown'), COALESCE($6, ''))
        ON CONFLICT (name) DO UPDATE
        SET source_type = EXCLUDED.source_type,
            url = EXCLUDED.url,
            cadence = EXCLUDED.cadence,
            reliability = EXCLUDED.reliability,
            notes = EXCLUDED.notes
        RETURNING
            source_id,
            name,
            source_type,
            url,
            cadence,
            reliability,
            notes,
            created_at::text AS created_at
        "#,
    )
    .bind(payload.name.trim().to_string())
    .bind(payload.source_type.trim().to_string())
    .bind(payload.url)
    .bind(payload.cadence)
    .bind(payload.reliability)
    .bind(payload.notes)
    .fetch_one(&state.pool)
    .await?;

    Ok(Json(source))
}

pub async fn list_ingestion_runs(
    State(state): State<AppState>,
    Query(query): Query<IngestionRunQuery>,
) -> ApiResult<Json<Vec<IngestionRun>>> {
    let runs = sqlx::query_as::<_, IngestionRun>(
        r#"
        SELECT
            r.run_id,
            r.source_id,
            s.name AS source_name,
            r.raw_artifact_id,
            r.status,
            r.started_at::text AS started_at,
            r.finished_at::text AS finished_at,
            r.raw_uri,
            r.row_count,
            r.error_message
        FROM audit.ingestion_runs r
        LEFT JOIN audit.data_sources s ON s.source_id = r.source_id
        WHERE ($1::bigint IS NULL OR r.source_id = $1)
        ORDER BY r.started_at DESC, r.run_id DESC
        LIMIT 100
        "#,
    )
    .bind(query.source_id)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(runs))
}

pub async fn create_ingestion_run(
    State(state): State<AppState>,
    Json(payload): Json<CreateIngestionRun>,
) -> ApiResult<Json<IngestionRun>> {
    payload
        .validate()
        .map_err(|message| ApiError::BadRequest(message.to_string()))?;

    let run = sqlx::query_as::<_, IngestionRun>(
        r#"
        INSERT INTO audit.ingestion_runs (
            source_id,
            status,
            raw_uri,
            raw_artifact_id,
            row_count,
            error_message
        )
        VALUES ($1, COALESCE($2, 'running'), $3, $4, $5, $6)
        RETURNING
            run_id,
            source_id,
            (
                SELECT name
                FROM audit.data_sources
                WHERE source_id = audit.ingestion_runs.source_id
            ) AS source_name,
            raw_artifact_id,
            status,
            started_at::text AS started_at,
            finished_at::text AS finished_at,
            raw_uri,
            row_count,
            error_message
        "#,
    )
    .bind(payload.source_id)
    .bind(payload.status)
    .bind(payload.raw_uri)
    .bind(payload.raw_artifact_id)
    .bind(payload.row_count)
    .bind(payload.error_message)
    .fetch_one(&state.pool)
    .await?;

    Ok(Json(run))
}

pub async fn finish_ingestion_run(
    State(state): State<AppState>,
    Path(run_id): Path<i64>,
    Json(payload): Json<FinishIngestionRun>,
) -> ApiResult<Json<IngestionRun>> {
    payload
        .validate()
        .map_err(|message| ApiError::BadRequest(message.to_string()))?;

    let run = sqlx::query_as::<_, IngestionRun>(
        r#"
        UPDATE audit.ingestion_runs
        SET status = $2,
            finished_at = now(),
            row_count = COALESCE($3, row_count),
            error_message = $4
        WHERE run_id = $1
        RETURNING
            run_id,
            source_id,
            (
                SELECT name
                FROM audit.data_sources
                WHERE source_id = audit.ingestion_runs.source_id
            ) AS source_name,
            raw_artifact_id,
            status,
            started_at::text AS started_at,
            finished_at::text AS finished_at,
            raw_uri,
            row_count,
            error_message
        "#,
    )
    .bind(run_id)
    .bind(payload.status)
    .bind(payload.row_count)
    .bind(payload.error_message)
    .fetch_optional(&state.pool)
    .await?
    .ok_or_else(|| ApiError::BadRequest("ingestion run not found".to_string()))?;

    Ok(Json(run))
}

pub async fn list_raw_artifacts(
    State(state): State<AppState>,
    Query(query): Query<RawArtifactQuery>,
) -> ApiResult<Json<Vec<RawArtifact>>> {
    if let Some(sha256) = query.sha256.as_deref() {
        validate_sha256(sha256)?;
    }

    let artifacts = sqlx::query_as::<_, RawArtifact>(
        r#"
        SELECT
            a.artifact_id,
            a.run_id,
            a.source_id,
            s.name AS source_name,
            a.original_filename,
            a.raw_uri,
            a.sha256,
            a.size_bytes,
            a.mime_type,
            a.uploaded_by,
            a.notes,
            a.created_at::text AS created_at
        FROM audit.raw_artifacts a
        LEFT JOIN audit.data_sources s ON s.source_id = a.source_id
        WHERE ($1::bigint IS NULL OR a.source_id = $1)
          AND (
              $2::bigint IS NULL
              OR a.run_id = $2
              OR EXISTS (
                  SELECT 1
                  FROM audit.ingestion_runs r
                  WHERE r.run_id = $2
                    AND r.raw_artifact_id = a.artifact_id
              )
          )
          AND ($3::text IS NULL OR a.sha256 = $3)
        ORDER BY a.created_at DESC, a.artifact_id DESC
        LIMIT 100
        "#,
    )
    .bind(query.source_id)
    .bind(query.run_id)
    .bind(query.sha256)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(artifacts))
}

pub async fn create_raw_artifact(
    State(state): State<AppState>,
    Json(payload): Json<CreateRawArtifact>,
) -> ApiResult<Json<RawArtifact>> {
    payload
        .validate()
        .map_err(|message| ApiError::BadRequest(message.to_string()))?;

    let mut tx = state.pool.begin().await?;
    let artifact = sqlx::query_as::<_, RawArtifact>(
        r#"
        INSERT INTO audit.raw_artifacts (
            run_id,
            source_id,
            original_filename,
            raw_uri,
            sha256,
            size_bytes,
            mime_type,
            uploaded_by,
            notes
        )
        VALUES (
            $1,
            COALESCE(
                $2,
                (
                    SELECT source_id
                    FROM audit.ingestion_runs
                    WHERE run_id = $1
                )
            ),
            $3,
            $4,
            $5,
            $6,
            $7,
            COALESCE($8, 'system'),
            COALESCE($9, '')
        )
        ON CONFLICT (sha256) DO UPDATE
        SET run_id = COALESCE(audit.raw_artifacts.run_id, EXCLUDED.run_id),
            source_id = COALESCE(audit.raw_artifacts.source_id, EXCLUDED.source_id),
            notes = CASE
                WHEN audit.raw_artifacts.notes = '' THEN EXCLUDED.notes
                ELSE audit.raw_artifacts.notes
            END
        RETURNING
            artifact_id,
            run_id,
            source_id,
            (
                SELECT name
                FROM audit.data_sources
                WHERE source_id = audit.raw_artifacts.source_id
            ) AS source_name,
            original_filename,
            raw_uri,
            sha256,
            size_bytes,
            mime_type,
            uploaded_by,
            notes,
            created_at::text AS created_at
        "#,
    )
    .bind(payload.run_id)
    .bind(payload.source_id)
    .bind(payload.original_filename.trim().to_string())
    .bind(payload.raw_uri.trim().to_string())
    .bind(payload.sha256.trim().to_string())
    .bind(payload.size_bytes)
    .bind(payload.mime_type)
    .bind(payload.uploaded_by)
    .bind(payload.notes)
    .fetch_one(&mut *tx)
    .await?;

    if let Some(run_id) = payload.run_id {
        sqlx::query(
            r#"
            UPDATE audit.ingestion_runs
            SET raw_artifact_id = $2,
                raw_uri = COALESCE(raw_uri, $3)
            WHERE run_id = $1
            "#,
        )
        .bind(run_id)
        .bind(artifact.artifact_id)
        .bind(&artifact.raw_uri)
        .execute(&mut *tx)
        .await?;
    }

    tx.commit().await?;
    Ok(Json(artifact))
}

pub async fn list_watchlist_items(
    State(state): State<AppState>,
    Query(query): Query<WatchlistQuery>,
) -> ApiResult<Json<Vec<WatchlistItem>>> {
    let items = sqlx::query_as::<_, WatchlistItem>(
        r#"
        SELECT
            watchlist_item_id,
            neighborhood_id,
            area_id,
            target_price_psm,
            status,
            label,
            priority,
            trigger_operator,
            trigger_price_psm,
            notes,
            created_at::text AS created_at,
            updated_at::text AS updated_at,
            last_reviewed_at::text AS last_reviewed_at
        FROM app.watchlist_items
        WHERE ($1::text IS NULL OR status = $1)
          AND ($1::text IS NOT NULL OR status <> 'archived')
          AND ($2::text IS NULL OR area_id = $2)
          AND ($3::text IS NULL OR neighborhood_id = $3)
          AND ($4::text IS NULL OR label = $4)
        ORDER BY updated_at DESC, watchlist_item_id DESC
        "#,
    )
    .bind(query.status)
    .bind(query.area_id)
    .bind(query.neighborhood_id)
    .bind(query.label)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(items))
}

pub async fn create_watchlist_item(
    State(state): State<AppState>,
    Json(payload): Json<CreateWatchlistItem>,
) -> ApiResult<Json<WatchlistItem>> {
    payload
        .validate()
        .map_err(|message| ApiError::BadRequest(message.to_string()))?;

    let mut tx = state.pool.begin().await?;
    let item = sqlx::query_as::<_, WatchlistItem>(
        r#"
        INSERT INTO app.watchlist_items (
            neighborhood_id,
            area_id,
            target_price_psm,
            label,
            priority,
            trigger_operator,
            trigger_price_psm,
            notes
        )
        VALUES (
            $1,
            $2,
            $3,
            COALESCE($4, 'general'),
            COALESCE($5, 'medium'),
            COALESCE($6, 'lte'),
            $7,
            COALESCE($8, '')
        )
        RETURNING
            watchlist_item_id,
            neighborhood_id,
            area_id,
            target_price_psm,
            status,
            label,
            priority,
            trigger_operator,
            trigger_price_psm,
            notes,
            created_at::text AS created_at,
            updated_at::text AS updated_at,
            last_reviewed_at::text AS last_reviewed_at
        "#,
    )
    .bind(payload.neighborhood_id)
    .bind(payload.area_id)
    .bind(payload.target_price_psm)
    .bind(payload.label)
    .bind(payload.priority)
    .bind(payload.trigger_operator)
    .bind(payload.trigger_price_psm)
    .bind(payload.notes)
    .fetch_one(&mut *tx)
    .await?;
    insert_watchlist_event(&mut tx, item.watchlist_item_id, "created", "创建观察标的").await?;
    tx.commit().await?;

    Ok(Json(item))
}

pub async fn update_watchlist_item(
    State(state): State<AppState>,
    Path(watchlist_item_id): Path<i64>,
    Json(payload): Json<UpdateWatchlistItem>,
) -> ApiResult<Json<WatchlistItem>> {
    payload
        .validate()
        .map_err(|message| ApiError::BadRequest(message.to_string()))?;

    let mut tx = state.pool.begin().await?;
    let item = sqlx::query_as::<_, WatchlistItem>(
        r#"
        UPDATE app.watchlist_items
        SET target_price_psm = COALESCE($2, target_price_psm),
            status = COALESCE($3, status),
            label = COALESCE($4, label),
            priority = COALESCE($5, priority),
            trigger_operator = COALESCE($6, trigger_operator),
            trigger_price_psm = COALESCE($7, trigger_price_psm),
            notes = COALESCE($8, notes),
            last_reviewed_at = CASE WHEN $9 THEN now() ELSE last_reviewed_at END,
            updated_at = now()
        WHERE watchlist_item_id = $1
        RETURNING
            watchlist_item_id,
            neighborhood_id,
            area_id,
            target_price_psm,
            status,
            label,
            priority,
            trigger_operator,
            trigger_price_psm,
            notes,
            created_at::text AS created_at,
            updated_at::text AS updated_at,
            last_reviewed_at::text AS last_reviewed_at
        "#,
    )
    .bind(watchlist_item_id)
    .bind(payload.target_price_psm)
    .bind(payload.status.clone())
    .bind(payload.label)
    .bind(payload.priority)
    .bind(payload.trigger_operator)
    .bind(payload.trigger_price_psm)
    .bind(payload.notes)
    .bind(payload.mark_reviewed.unwrap_or(false))
    .fetch_optional(&mut *tx)
    .await?
    .ok_or_else(|| ApiError::BadRequest("watchlist item not found".to_string()))?;
    insert_watchlist_event(
        &mut tx,
        item.watchlist_item_id,
        if payload.mark_reviewed.unwrap_or(false) {
            "reviewed"
        } else {
            "updated"
        },
        if payload.mark_reviewed.unwrap_or(false) {
            "完成一次人工复核"
        } else {
            "更新观察标的"
        },
    )
    .await?;
    tx.commit().await?;

    Ok(Json(item))
}

pub async fn archive_watchlist_item(
    State(state): State<AppState>,
    Path(watchlist_item_id): Path<i64>,
) -> ApiResult<Json<WatchlistItem>> {
    let mut tx = state.pool.begin().await?;
    let item = sqlx::query_as::<_, WatchlistItem>(
        r#"
        UPDATE app.watchlist_items
        SET status = 'archived',
            updated_at = now()
        WHERE watchlist_item_id = $1
        RETURNING
            watchlist_item_id,
            neighborhood_id,
            area_id,
            target_price_psm,
            status,
            label,
            priority,
            trigger_operator,
            trigger_price_psm,
            notes,
            created_at::text AS created_at,
            updated_at::text AS updated_at,
            last_reviewed_at::text AS last_reviewed_at
        "#,
    )
    .bind(watchlist_item_id)
    .fetch_optional(&mut *tx)
    .await?
    .ok_or_else(|| ApiError::BadRequest("watchlist item not found".to_string()))?;
    insert_watchlist_event(&mut tx, item.watchlist_item_id, "archived", "归档观察标的").await?;
    tx.commit().await?;

    Ok(Json(item))
}

pub async fn list_watchlist_events(
    State(state): State<AppState>,
    Path(watchlist_item_id): Path<i64>,
) -> ApiResult<Json<Vec<WatchlistEvent>>> {
    let events = sqlx::query_as::<_, WatchlistEvent>(
        r#"
        SELECT
            event_id,
            watchlist_item_id,
            event_type,
            summary,
            created_at::text AS created_at
        FROM app.watchlist_events
        WHERE watchlist_item_id = $1
        ORDER BY created_at DESC, event_id DESC
        LIMIT 50
        "#,
    )
    .bind(watchlist_item_id)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(events))
}

pub async fn create_watchlist_event(
    State(state): State<AppState>,
    Path(watchlist_item_id): Path<i64>,
    Json(payload): Json<CreateWatchlistEvent>,
) -> ApiResult<Json<WatchlistEvent>> {
    payload
        .validate()
        .map_err(|message| ApiError::BadRequest(message.to_string()))?;

    let event = sqlx::query_as::<_, WatchlistEvent>(
        r#"
        INSERT INTO app.watchlist_events (
            watchlist_item_id,
            event_type,
            summary
        )
        VALUES ($1, $2, $3)
        RETURNING
            event_id,
            watchlist_item_id,
            event_type,
            summary,
            created_at::text AS created_at
        "#,
    )
    .bind(watchlist_item_id)
    .bind(payload.event_type.trim().to_string())
    .bind(payload.summary.trim().to_string())
    .fetch_one(&state.pool)
    .await?;

    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 }))
}

pub async fn list_reports(
    State(state): State<AppState>,
    Query(query): Query<ReportQuery>,
) -> ApiResult<Json<Vec<Report>>> {
    validate_report_query(&query)?;
    let reports = sqlx::query_as::<_, Report>(
        report_select_sql(
            r#"
            WHERE ($1::text IS NULL OR r.report_type = $1)
              AND ($2::text IS NULL OR r.target_month = $2)
              AND ($3::text IS NULL OR r.area_id = $3)
              AND ($4::text IS NULL OR r.neighborhood_id = $4)
              AND ($5::text IS NULL OR r.status = $5)
            ORDER BY r.updated_at DESC, r.report_id DESC
            LIMIT 80
            "#,
        )
        .as_str(),
    )
    .bind(query.report_type)
    .bind(query.target_month)
    .bind(query.area_id)
    .bind(query.neighborhood_id)
    .bind(query.status)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(reports))
}

pub async fn get_report(
    State(state): State<AppState>,
    Path(report_id): Path<i64>,
) -> ApiResult<Json<Report>> {
    let report = sqlx::query_as::<_, Report>(report_select_sql("WHERE r.report_id = $1").as_str())
        .bind(report_id)
        .fetch_optional(&state.pool)
        .await?
        .ok_or_else(|| ApiError::BadRequest("report not found".to_string()))?;

    Ok(Json(report))
}

pub async fn create_report(
    State(state): State<AppState>,
    Json(payload): Json<CreateReport>,
) -> ApiResult<Json<Report>> {
    validate_create_report(&payload)?;
    let linked_metrics = payload
        .linked_metrics
        .clone()
        .unwrap_or_else(|| serde_json::json!({}));
    if !linked_metrics.is_object() {
        return Err(ApiError::BadRequest(
            "linked_metrics must be a JSON object".to_string(),
        ));
    }

    let mut tx = state.pool.begin().await?;
    let report = insert_report_from_payload(
        &mut tx,
        CreateReport {
            linked_metrics: Some(linked_metrics),
            ..payload
        },
    )
    .await?;
    tx.commit().await?;

    Ok(Json(report))
}

pub async fn list_report_templates(
    State(state): State<AppState>,
) -> ApiResult<Json<Vec<ReportTemplate>>> {
    let templates = sqlx::query_as::<_, ReportTemplate>(
        r#"
        SELECT
            template_id,
            template_key,
            name,
            report_type,
            cadence,
            description,
            body_template,
            created_at::text AS created_at,
            updated_at::text AS updated_at
        FROM app.report_templates
        ORDER BY cadence, template_key
        "#,
    )
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(templates))
}

pub async fn list_report_generation_runs(
    State(state): State<AppState>,
    Query(query): Query<ReportGenerationRunQuery>,
) -> ApiResult<Json<Vec<ReportGenerationRun>>> {
    validate_report_generation_query(&query)?;
    let runs = sqlx::query_as::<_, ReportGenerationRun>(
        r#"
        SELECT
            r.generation_run_id,
            r.template_id,
            t.template_key,
            t.name AS template_name,
            r.target_month,
            r.status,
            r.report_id,
            p.title AS report_title,
            r.parameters,
            r.error_message,
            r.started_at::text AS started_at,
            r.finished_at::text AS finished_at
        FROM app.report_generation_runs r
        JOIN app.report_templates t ON t.template_id = r.template_id
        LEFT JOIN app.reports p ON p.report_id = r.report_id
        WHERE ($1::text IS NULL OR r.target_month = $1)
          AND ($2::text IS NULL OR t.template_key = $2)
          AND ($3::text IS NULL OR r.status = $3)
        ORDER BY r.generation_run_id DESC
        LIMIT 80
        "#,
    )
    .bind(query.target_month)
    .bind(query.template_key)
    .bind(query.status)
    .fetch_all(&state.pool)
    .await?;

    Ok(Json(runs))
}

pub async fn create_report_generation_run(
    State(state): State<AppState>,
    Json(payload): Json<CreateReportGenerationRun>,
) -> ApiResult<Json<ReportGenerationRunResponse>> {
    validate_month(&payload.target_month)?;
    payload
        .validate()
        .map_err(|message| ApiError::BadRequest(message.to_string()))?;
    validate_identifier(&payload.template_key)?;
    let parameters = payload
        .parameters
        .clone()
        .unwrap_or_else(|| serde_json::json!({}));

    let mut tx = state.pool.begin().await?;
    let template = fetch_report_template(&mut tx, &payload.template_key)
        .await?
        .ok_or_else(|| ApiError::BadRequest("report template not found".to_string()))?;
    let generation_run_id = sqlx::query_scalar::<_, i64>(
        r#"
        INSERT INTO app.report_generation_runs (
            template_id,
            target_month,
            status,
            parameters
        )
        VALUES ($1, $2, 'running', $3)
        RETURNING generation_run_id
        "#,
    )
    .bind(template.template_id)
    .bind(payload.target_month.trim().to_string())
    .bind(parameters.clone())
    .fetch_one(&mut *tx)
    .await?;

    let generated =
        build_generated_report(&mut tx, &template, &payload.target_month, &parameters).await;
    match generated {
        Ok(create_report_payload) => {
            let report = insert_report_from_payload(&mut tx, create_report_payload).await?;
            finish_report_generation_run(
                &mut tx,
                generation_run_id,
                "succeeded",
                Some(report.report_id),
                None,
            )
            .await?;
            let generation_run = fetch_report_generation_run_by_id(&mut tx, generation_run_id)
                .await?
                .ok_or_else(|| ApiError::BadRequest("generation run not found".to_string()))?;
            tx.commit().await?;
            Ok(Json(ReportGenerationRunResponse {
                generation_run,
                report: Some(report),
            }))
        }
        Err(message) => {
            finish_report_generation_run(
                &mut tx,
                generation_run_id,
                "failed",
                None,
                Some(&message),
            )
            .await?;
            let generation_run = fetch_report_generation_run_by_id(&mut tx, generation_run_id)
                .await?
                .ok_or_else(|| ApiError::BadRequest("generation run not found".to_string()))?;
            tx.commit().await?;
            Ok(Json(ReportGenerationRunResponse {
                generation_run,
                report: None,
            }))
        }
    }
}

async fn insert_watchlist_event(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    watchlist_item_id: i64,
    event_type: &str,
    summary: &str,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        INSERT INTO app.watchlist_events (
            watchlist_item_id,
            event_type,
            summary
        )
        VALUES ($1, $2, $3)
        "#,
    )
    .bind(watchlist_item_id)
    .bind(event_type)
    .bind(summary)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

fn report_select_sql(where_clause: &str) -> String {
    format!(
        r#"
        SELECT
            r.report_id,
            r.report_type,
            r.title,
            r.target_month,
            r.area_id,
            a.name AS area_name,
            r.neighborhood_id,
            n.name AS neighborhood_name,
            r.summary,
            r.content_markdown,
            r.linked_metrics,
            r.status,
            r.created_at::text AS created_at,
            r.updated_at::text AS updated_at
        FROM app.reports r
        LEFT JOIN silver.areas a ON a.area_id = r.area_id
        LEFT JOIN silver.neighborhoods n ON n.neighborhood_id = r.neighborhood_id
        {where_clause}
        "#
    )
}

async fn fetch_report_by_id(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    report_id: i64,
) -> Result<Option<Report>, sqlx::Error> {
    sqlx::query_as::<_, Report>(report_select_sql("WHERE r.report_id = $1").as_str())
        .bind(report_id)
        .fetch_optional(&mut **tx)
        .await
}

async fn fetch_report_template(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    template_key: &str,
) -> Result<Option<ReportTemplate>, sqlx::Error> {
    sqlx::query_as::<_, ReportTemplate>(
        r#"
        SELECT
            template_id,
            template_key,
            name,
            report_type,
            cadence,
            description,
            body_template,
            created_at::text AS created_at,
            updated_at::text AS updated_at
        FROM app.report_templates
        WHERE template_key = $1
        "#,
    )
    .bind(template_key)
    .fetch_optional(&mut **tx)
    .await
}

async fn fetch_report_generation_run_by_id(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    generation_run_id: i64,
) -> Result<Option<ReportGenerationRun>, sqlx::Error> {
    sqlx::query_as::<_, ReportGenerationRun>(
        r#"
        SELECT
            r.generation_run_id,
            r.template_id,
            t.template_key,
            t.name AS template_name,
            r.target_month,
            r.status,
            r.report_id,
            p.title AS report_title,
            r.parameters,
            r.error_message,
            r.started_at::text AS started_at,
            r.finished_at::text AS finished_at
        FROM app.report_generation_runs r
        JOIN app.report_templates t ON t.template_id = r.template_id
        LEFT JOIN app.reports p ON p.report_id = r.report_id
        WHERE r.generation_run_id = $1
        "#,
    )
    .bind(generation_run_id)
    .fetch_optional(&mut **tx)
    .await
}

async fn finish_report_generation_run(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    generation_run_id: i64,
    status: &str,
    report_id: Option<i64>,
    error_message: Option<&str>,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE app.report_generation_runs
        SET status = $2,
            report_id = $3,
            error_message = $4,
            finished_at = now()
        WHERE generation_run_id = $1
        "#,
    )
    .bind(generation_run_id)
    .bind(status)
    .bind(report_id)
    .bind(error_message)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

async fn insert_report_from_payload(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    payload: CreateReport,
) -> Result<Report, sqlx::Error> {
    let report_id = sqlx::query_scalar::<_, i64>(
        r#"
        INSERT INTO app.reports (
            report_type,
            title,
            target_month,
            area_id,
            neighborhood_id,
            summary,
            content_markdown,
            linked_metrics,
            status
        )
        VALUES ($1, $2, $3, $4, $5, COALESCE($6, ''), $7, $8, COALESCE($9, 'draft'))
        RETURNING report_id
        "#,
    )
    .bind(payload.report_type.trim().to_string())
    .bind(payload.title.trim().to_string())
    .bind(payload.target_month.clone())
    .bind(payload.area_id.clone())
    .bind(payload.neighborhood_id.clone())
    .bind(payload.summary.clone())
    .bind(payload.content_markdown.trim().to_string())
    .bind(
        payload
            .linked_metrics
            .unwrap_or_else(|| serde_json::json!({})),
    )
    .bind(payload.status.clone())
    .fetch_one(&mut **tx)
    .await?;

    fetch_report_by_id(tx, report_id)
        .await?
        .ok_or(sqlx::Error::RowNotFound)
}

async fn build_generated_report(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    template: &ReportTemplate,
    target_month: &str,
    parameters: &serde_json::Value,
) -> Result<CreateReport, String> {
    match template.template_key.as_str() {
        "monthly_market" => {
            build_monthly_market_report(tx, template, target_month, parameters).await
        }
        "weekly_watchlist" => {
            build_weekly_watchlist_report(tx, template, target_month, parameters).await
        }
        other => Err(format!("template '{other}' is not supported by generator")),
    }
}

async fn build_monthly_market_report(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    template: &ReportTemplate,
    target_month: &str,
    parameters: &serde_json::Value,
) -> Result<CreateReport, String> {
    let scores = fetch_area_scores_in_tx(tx, target_month)
        .await
        .map_err(|error| format!("failed to fetch area scores: {error}"))?;
    if scores.is_empty() {
        return Err(format!("no area scores found for month {target_month}"));
    }
    let top = scores
        .first()
        .ok_or_else(|| "no top area found".to_string())?;
    let weakest = scores
        .last()
        .ok_or_else(|| "no weakest area found".to_string())?;
    let high_supply = scores
        .iter()
        .max_by(|left, right| {
            left.supply_risk_score
                .partial_cmp(&right.supply_risk_score)
                .unwrap_or(std::cmp::Ordering::Equal)
        })
        .ok_or_else(|| "no supply risk area found".to_string())?;
    let avg_score = average(scores.iter().map(|score| score.investment_score));
    let avg_yield = average(scores.iter().map(|score| score.annual_rent_yield_pct));
    let market_summary = format!(
        "## 核心结论\n\n- 综合评分最高板块：{}（{}，{}）。\n- 综合评分最低板块：{}（{}，{}）。\n- 供应压力最高板块：{}（供应风险 {}）。\n- 样本平均投资观察评分：{}；样本平均年化租金收益率：{}%。",
        top.name,
        format_score(top.investment_score),
        top.recommendation,
        weakest.name,
        format_score(weakest.investment_score),
        weakest.recommendation,
        high_supply.name,
        format_score(high_supply.supply_risk_score),
        format_score(avg_score),
        round_to(avg_yield, 2)
    );
    let area_table = monthly_area_table(&scores);
    let risk_section = format!(
        "## 风险解释\n\n- {} 的供应风险最高，需要跟踪新增供应和挂牌去化。\n- {} 综合评分最低，进入谨慎复核列表。\n- 自动报告只汇总结构化指标，具体买卖动作仍需人工复核。",
        high_supply.name, weakest.name
    );
    let next_actions = "## 下月跟踪\n\n- 复核高分板块内小区分化。\n- 跟踪观察池价格触发和成交量变化。\n- 对政策和利率变化进行情景推演。";
    let content = template
        .body_template
        .replace("{{month}}", target_month)
        .replace("{{market_summary}}", &market_summary)
        .replace("{{area_table}}", &area_table)
        .replace("{{risk_section}}", &risk_section)
        .replace("{{next_actions}}", next_actions);

    Ok(CreateReport {
        report_type: template.report_type.clone(),
        title: format!("上海房市投资研究月报 {target_month}"),
        target_month: Some(target_month.to_string()),
        area_id: None,
        neighborhood_id: None,
        summary: Some(format!(
            "最高评分 {}，最低评分 {}，供应压力最高 {}。",
            top.name, weakest.name, high_supply.name
        )),
        content_markdown: content,
        linked_metrics: Some(serde_json::json!({
            "template_key": template.template_key,
            "month": target_month,
            "top_area_id": top.area_id,
            "weakest_area_id": weakest.area_id,
            "highest_supply_risk_area_id": high_supply.area_id,
            "area_count": scores.len(),
            "parameters": parameters
        })),
        status: Some("published".to_string()),
    })
}

async fn build_weekly_watchlist_report(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    template: &ReportTemplate,
    target_month: &str,
    parameters: &serde_json::Value,
) -> Result<CreateReport, String> {
    let items = fetch_watchlist_items_in_tx(tx)
        .await
        .map_err(|error| format!("failed to fetch watchlist items: {error}"))?;
    let active_count = items.iter().filter(|item| item.status == "active").count();
    let paused_count = items.iter().filter(|item| item.status == "paused").count();
    let high_priority_count = items.iter().filter(|item| item.priority == "high").count();
    let watchlist_summary = format!(
        "## 观察池概览\n\n- 当前观察标的 {} 个，其中 active {} 个、paused {} 个。\n- 高优先级标的 {} 个。\n- 本周重点复核价格触发、成交量变化和最新备注。",
        items.len(),
        active_count,
        paused_count,
        high_priority_count
    );
    let watchlist_table = watchlist_report_table(&items);
    let next_actions = "## 下一步动作\n\n- 对 high 优先级标的补充人工复核记录。\n- 检查触发价附近标的是否需要调整观察价。\n- 将异常价格变化同步到预警模块。";
    let content = template
        .body_template
        .replace("{{month}}", target_month)
        .replace("{{watchlist_summary}}", &watchlist_summary)
        .replace("{{watchlist_table}}", &watchlist_table)
        .replace("{{next_actions}}", next_actions);

    Ok(CreateReport {
        report_type: template.report_type.clone(),
        title: format!("上海房市观察池周报 {target_month}"),
        target_month: Some(target_month.to_string()),
        area_id: None,
        neighborhood_id: None,
        summary: Some(format!(
            "观察标的 {} 个，高优先级 {} 个。",
            items.len(),
            high_priority_count
        )),
        content_markdown: content,
        linked_metrics: Some(serde_json::json!({
            "template_key": template.template_key,
            "month": target_month,
            "watchlist_count": items.len(),
            "active_count": active_count,
            "high_priority_count": high_priority_count,
            "parameters": parameters
        })),
        status: Some("published".to_string()),
    })
}

async fn fetch_area_scores_in_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    month: &str,
) -> Result<Vec<AreaScore>, sqlx::Error> {
    sqlx::query_as::<_, AreaScore>(
        r#"
        SELECT
            area_id,
            name,
            district,
            segment,
            month,
            investment_score,
            recommendation,
            liquidity_score,
            momentum_score,
            rent_support_score,
            safety_margin_score,
            credit_support_score,
            supply_risk_score,
            valuation_pressure_score,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            rent_price_psm,
            median_days_on_market,
            annual_rent_yield_pct,
            listing_pressure_ratio,
            discount_pct,
            price_momentum_pct,
            volume_momentum_pct
        FROM gold.area_scores
        WHERE month = $1
        ORDER BY investment_score DESC, area_id ASC
        "#,
    )
    .bind(month)
    .fetch_all(&mut **tx)
    .await
}

async fn fetch_watchlist_items_in_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> Result<Vec<WatchlistItem>, sqlx::Error> {
    sqlx::query_as::<_, WatchlistItem>(
        r#"
        SELECT
            watchlist_item_id,
            neighborhood_id,
            area_id,
            target_price_psm,
            status,
            label,
            priority,
            trigger_operator,
            trigger_price_psm,
            notes,
            created_at::text AS created_at,
            updated_at::text AS updated_at,
            last_reviewed_at::text AS last_reviewed_at
        FROM app.watchlist_items
        WHERE status <> 'archived'
        ORDER BY priority DESC, updated_at DESC, watchlist_item_id DESC
        LIMIT 100
        "#,
    )
    .fetch_all(&mut **tx)
    .await
}

fn monthly_area_table(scores: &[AreaScore]) -> String {
    let mut lines = vec![
        "## 板块评分".to_string(),
        String::new(),
        "| 排名 | 板块 | 行政区 | 综合分 | 建议 | 成交均价/㎡ | 成交套数 | 供应风险 |".to_string(),
        "| --- | --- | --- | ---: | --- | ---: | ---: | ---: |".to_string(),
    ];
    for (index, score) in scores.iter().enumerate() {
        lines.push(format!(
            "| {} | {} | {} | {} | {} | {:.0} | {} | {} |",
            index + 1,
            score.name,
            score.district,
            format_score(score.investment_score),
            score.recommendation,
            score.transaction_price_psm,
            score.transaction_count,
            format_score(score.supply_risk_score)
        ));
    }
    lines.join("\n")
}

fn watchlist_report_table(items: &[WatchlistItem]) -> String {
    let mut lines = vec![
        "## 观察池明细".to_string(),
        String::new(),
        "| ID | 类型 | 状态 | 优先级 | 目标价/㎡ | 触发价/㎡ | 标签 |".to_string(),
        "| ---: | --- | --- | --- | ---: | ---: | --- |".to_string(),
    ];
    for item in items.iter().take(30) {
        lines.push(format!(
            "| {} | {} | {} | {} | {} | {} | {} |",
            item.watchlist_item_id,
            item.neighborhood_id
                .as_deref()
                .or(item.area_id.as_deref())
                .unwrap_or("-"),
            item.status,
            item.priority,
            format_optional_price(item.target_price_psm),
            format_optional_price(item.trigger_price_psm),
            item.label
        ));
    }
    lines.join("\n")
}

fn format_score(value: f64) -> String {
    format!("{:.1}", round_to(value, 1))
}

fn format_optional_price(value: Option<f64>) -> String {
    value
        .map(|item| format!("{:.0}", item))
        .unwrap_or_else(|| "-".to_string())
}

fn validate_report_query(query: &ReportQuery) -> ApiResult<()> {
    if let Some(report_type) = query.report_type.as_deref() {
        validate_report_type(report_type)?;
    }
    if let Some(status) = query.status.as_deref() {
        validate_report_status(status)?;
    }
    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)?;
    }
    if let Some(neighborhood_id) = query.neighborhood_id.as_deref() {
        validate_identifier(neighborhood_id)?;
    }
    Ok(())
}

fn validate_create_report(payload: &CreateReport) -> ApiResult<()> {
    payload
        .validate()
        .map_err(|message| ApiError::BadRequest(message.to_string()))?;
    validate_report_type(&payload.report_type)?;
    if let Some(status) = payload.status.as_deref() {
        validate_report_status(status)?;
    }
    if let Some(month) = payload.target_month.as_deref() {
        validate_month(month)?;
    }
    if let Some(area_id) = payload.area_id.as_deref() {
        validate_identifier(area_id)?;
    }
    if let Some(neighborhood_id) = payload.neighborhood_id.as_deref() {
        validate_identifier(neighborhood_id)?;
    }
    Ok(())
}

fn validate_report_generation_query(query: &ReportGenerationRunQuery) -> ApiResult<()> {
    if let Some(month) = query.target_month.as_deref() {
        validate_month(month)?;
    }
    if let Some(template_key) = query.template_key.as_deref() {
        validate_identifier(template_key)?;
    }
    if let Some(status) = query.status.as_deref() {
        validate_generation_status(status)?;
    }
    Ok(())
}

async fn fetch_area_scores(state: &AppState, month: &str) -> Result<Vec<AreaScore>, sqlx::Error> {
    sqlx::query_as::<_, AreaScore>(
        r#"
        SELECT
            area_id,
            name,
            district,
            segment,
            month,
            investment_score,
            recommendation,
            liquidity_score,
            momentum_score,
            rent_support_score,
            safety_margin_score,
            credit_support_score,
            supply_risk_score,
            valuation_pressure_score,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            rent_price_psm,
            median_days_on_market,
            annual_rent_yield_pct,
            listing_pressure_ratio,
            discount_pct,
            price_momentum_pct,
            volume_momentum_pct
        FROM gold.area_scores
        WHERE month = $1
        ORDER BY investment_score DESC, area_id ASC
        "#,
    )
    .bind(month)
    .fetch_all(&state.pool)
    .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,
    model_version: &str,
    model_run_id: i64,
) -> Result<i32, String> {
    let current_rows = fetch_area_metric_rows(tx, 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 {month}"));
    }

    let prior_month = previous_month(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, month)
        .await
        .map_err(|error| format!("failed to fetch area price history: {error}"))?;
    let scores = compute_area_scores(&current_rows, &prior_rows, &price_history);

    for score in &scores {
        upsert_area_score(tx, score, model_version, model_run_id)
            .await
            .map_err(|error| format!("failed to upsert area score {}: {error}", score.area_id))?;
    }

    Ok(scores.len() as i32)
}

async fn fetch_area_metric_rows(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    month: &str,
) -> Result<Vec<AreaMetricRow>, sqlx::Error> {
    sqlx::query_as::<_, AreaMetricRow>(
        r#"
        SELECT
            a.area_id,
            a.name,
            a.district,
            a.segment,
            m.month,
            m.transaction_count,
            m.transaction_price_psm,
            m.listing_count,
            m.listing_price_psm,
            m.median_days_on_market,
            m.rent_price_psm,
            m.new_supply_units,
            m.mortgage_rate_pct,
            m.policy_signal
        FROM silver.area_monthly_metrics m
        JOIN silver.areas a ON a.area_id = m.area_id
        WHERE m.month = $1
        ORDER BY a.district, a.name
        "#,
    )
    .bind(month)
    .fetch_all(&mut **tx)
    .await
}

async fn fetch_area_price_history(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    through_month: &str,
) -> Result<std::collections::HashMap<String, Vec<f64>>, sqlx::Error> {
    let rows = sqlx::query_as::<_, (String, f64)>(
        r#"
        SELECT area_id, transaction_price_psm
        FROM silver.area_monthly_metrics
        WHERE month <= $1
        ORDER BY area_id, month
        "#,
    )
    .bind(through_month)
    .fetch_all(&mut **tx)
    .await?;

    let mut history = std::collections::HashMap::<String, Vec<f64>>::new();
    for (area_id, price) in rows {
        history.entry(area_id).or_default().push(price);
    }
    Ok(history)
}

async fn upsert_area_score(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    score: &ComputedAreaScore,
    model_version: &str,
    model_run_id: i64,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        INSERT INTO gold.area_scores (
            area_id,
            name,
            district,
            segment,
            month,
            investment_score,
            recommendation,
            liquidity_score,
            momentum_score,
            rent_support_score,
            safety_margin_score,
            credit_support_score,
            supply_risk_score,
            valuation_pressure_score,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            rent_price_psm,
            median_days_on_market,
            annual_rent_yield_pct,
            listing_pressure_ratio,
            discount_pct,
            price_momentum_pct,
            volume_momentum_pct,
            model_version,
            model_run_id
        )
        VALUES (
            $1, $2, $3, $4, $5,
            $6, $7, $8, $9, $10,
            $11, $12, $13, $14, $15,
            $16, $17, $18, $19, $20,
            $21, $22, $23, $24, $25,
            $26, $27
        )
        ON CONFLICT (area_id, month) DO UPDATE
        SET name = EXCLUDED.name,
            district = EXCLUDED.district,
            segment = EXCLUDED.segment,
            investment_score = EXCLUDED.investment_score,
            recommendation = EXCLUDED.recommendation,
            liquidity_score = EXCLUDED.liquidity_score,
            momentum_score = EXCLUDED.momentum_score,
            rent_support_score = EXCLUDED.rent_support_score,
            safety_margin_score = EXCLUDED.safety_margin_score,
            credit_support_score = EXCLUDED.credit_support_score,
            supply_risk_score = EXCLUDED.supply_risk_score,
            valuation_pressure_score = EXCLUDED.valuation_pressure_score,
            transaction_count = EXCLUDED.transaction_count,
            transaction_price_psm = EXCLUDED.transaction_price_psm,
            listing_count = EXCLUDED.listing_count,
            listing_price_psm = EXCLUDED.listing_price_psm,
            rent_price_psm = EXCLUDED.rent_price_psm,
            median_days_on_market = EXCLUDED.median_days_on_market,
            annual_rent_yield_pct = EXCLUDED.annual_rent_yield_pct,
            listing_pressure_ratio = EXCLUDED.listing_pressure_ratio,
            discount_pct = EXCLUDED.discount_pct,
            price_momentum_pct = EXCLUDED.price_momentum_pct,
            volume_momentum_pct = EXCLUDED.volume_momentum_pct,
            model_version = EXCLUDED.model_version,
            model_run_id = EXCLUDED.model_run_id,
            computed_at = now()
        "#,
    )
    .bind(&score.area_id)
    .bind(&score.name)
    .bind(&score.district)
    .bind(&score.segment)
    .bind(&score.month)
    .bind(score.investment_score)
    .bind(&score.recommendation)
    .bind(score.liquidity_score)
    .bind(score.momentum_score)
    .bind(score.rent_support_score)
    .bind(score.safety_margin_score)
    .bind(score.credit_support_score)
    .bind(score.supply_risk_score)
    .bind(score.valuation_pressure_score)
    .bind(score.transaction_count)
    .bind(score.transaction_price_psm)
    .bind(score.listing_count)
    .bind(score.listing_price_psm)
    .bind(score.rent_price_psm)
    .bind(score.median_days_on_market)
    .bind(score.annual_rent_yield_pct)
    .bind(score.listing_pressure_ratio)
    .bind(score.discount_pct)
    .bind(score.price_momentum_pct)
    .bind(score.volume_momentum_pct)
    .bind(model_version)
    .bind(model_run_id)
    .execute(&mut **tx)
    .await?;

    sqlx::query(
        r#"
        INSERT INTO gold.area_score_snapshots (
            model_run_id,
            area_id,
            name,
            district,
            segment,
            month,
            investment_score,
            recommendation,
            liquidity_score,
            momentum_score,
            rent_support_score,
            safety_margin_score,
            credit_support_score,
            supply_risk_score,
            valuation_pressure_score,
            transaction_count,
            transaction_price_psm,
            listing_count,
            listing_price_psm,
            rent_price_psm,
            median_days_on_market,
            annual_rent_yield_pct,
            listing_pressure_ratio,
            discount_pct,
            price_momentum_pct,
            volume_momentum_pct,
            model_version
        )
        VALUES (
            $1, $2, $3, $4, $5,
            $6, $7, $8, $9, $10,
            $11, $12, $13, $14, $15,
            $16, $17, $18, $19, $20,
            $21, $22, $23, $24, $25,
            $26, $27
        )
        ON CONFLICT (model_run_id, area_id) DO UPDATE
        SET name = EXCLUDED.name,
            district = EXCLUDED.district,
            segment = EXCLUDED.segment,
            investment_score = EXCLUDED.investment_score,
            recommendation = EXCLUDED.recommendation,
            liquidity_score = EXCLUDED.liquidity_score,
            momentum_score = EXCLUDED.momentum_score,
            rent_support_score = EXCLUDED.rent_support_score,
            safety_margin_score = EXCLUDED.safety_margin_score,
            credit_support_score = EXCLUDED.credit_support_score,
            supply_risk_score = EXCLUDED.supply_risk_score,
            valuation_pressure_score = EXCLUDED.valuation_pressure_score,
            transaction_count = EXCLUDED.transaction_count,
            transaction_price_psm = EXCLUDED.transaction_price_psm,
            listing_count = EXCLUDED.listing_count,
            listing_price_psm = EXCLUDED.listing_price_psm,
            rent_price_psm = EXCLUDED.rent_price_psm,
            median_days_on_market = EXCLUDED.median_days_on_market,
            annual_rent_yield_pct = EXCLUDED.annual_rent_yield_pct,
            listing_pressure_ratio = EXCLUDED.listing_pressure_ratio,
            discount_pct = EXCLUDED.discount_pct,
            price_momentum_pct = EXCLUDED.price_momentum_pct,
            volume_momentum_pct = EXCLUDED.volume_momentum_pct,
            model_version = EXCLUDED.model_version,
            computed_at = now()
        "#,
    )
    .bind(model_run_id)
    .bind(&score.area_id)
    .bind(&score.name)
    .bind(&score.district)
    .bind(&score.segment)
    .bind(&score.month)
    .bind(score.investment_score)
    .bind(&score.recommendation)
    .bind(score.liquidity_score)
    .bind(score.momentum_score)
    .bind(score.rent_support_score)
    .bind(score.safety_margin_score)
    .bind(score.credit_support_score)
    .bind(score.supply_risk_score)
    .bind(score.valuation_pressure_score)
    .bind(score.transaction_count)
    .bind(score.transaction_price_psm)
    .bind(score.listing_count)
    .bind(score.listing_price_psm)
    .bind(score.rent_price_psm)
    .bind(score.median_days_on_market)
    .bind(score.annual_rent_yield_pct)
    .bind(score.listing_pressure_ratio)
    .bind(score.discount_pct)
    .bind(score.price_momentum_pct)
    .bind(score.volume_momentum_pct)
    .bind(model_version)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

async fn finish_model_run(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    model_run_id: i64,
    status: &str,
    row_count: Option<i32>,
    error_message: Option<&str>,
) -> Result<ModelRun, sqlx::Error> {
    sqlx::query_as::<_, ModelRun>(
        r#"
        UPDATE gold.model_runs
        SET status = $2,
            finished_at = now(),
            row_count = COALESCE($3, row_count),
            error_message = $4
        WHERE model_run_id = $1
        RETURNING
            model_run_id,
            model_name,
            model_version,
            target_month,
            status,
            parameters,
            started_at::text AS started_at,
            finished_at::text AS finished_at,
            row_count,
            error_message
        "#,
    )
    .bind(model_run_id)
    .bind(status)
    .bind(row_count)
    .bind(error_message)
    .fetch_one(&mut **tx)
    .await
}

async fn fetch_model_run(
    state: &AppState,
    model_run_id: i64,
) -> Result<Option<ModelRun>, sqlx::Error> {
    sqlx::query_as::<_, ModelRun>(
        r#"
        SELECT
            model_run_id,
            model_name,
            model_version,
            target_month,
            status,
            parameters,
            started_at::text AS started_at,
            finished_at::text AS finished_at,
            row_count,
            error_message
        FROM gold.model_runs
        WHERE model_run_id = $1
        "#,
    )
    .bind(model_run_id)
    .fetch_optional(&state.pool)
    .await
}

fn average(values: impl Iterator<Item = f64>) -> f64 {
    let (sum, count) = values.fold((0.0, 0usize), |(sum, count), value| {
        (sum + value, count + 1)
    });
    if count == 0 {
        0.0
    } else {
        (sum / count as f64 * 10.0).round() / 10.0
    }
}

fn diagnostic_metric(
    key: &str,
    label: &str,
    unit: &str,
    current_value: f64,
    values: &[f64],
) -> DiagnosticMetric {
    let sample_size = values.len();
    let historical_min = values.iter().copied().fold(f64::INFINITY, f64::min);
    let historical_max = values.iter().copied().fold(f64::NEG_INFINITY, f64::max);
    let percentile = percentile_rank(current_value, values);
    let anomaly = sample_size >= 3 && !(10.0..=90.0).contains(&percentile);
    let anomaly_direction = if !anomaly {
        None
    } else if percentile > 90.0 {
        Some("high".to_string())
    } else if percentile < 10.0 {
        Some("low".to_string())
    } else {
        None
    };
    let severity = if sample_size < 3 {
        "insufficient_history"
    } else if !(5.0..=95.0).contains(&percentile) {
        "high"
    } else if anomaly {
        "medium"
    } else {
        "normal"
    };

    DiagnosticMetric {
        key: key.to_string(),
        label: label.to_string(),
        unit: unit.to_string(),
        current_value: round_to(current_value, 2),
        percentile: round_to(percentile, 1),
        historical_min: round_to(historical_min, 2),
        historical_max: round_to(historical_max, 2),
        sample_size,
        anomaly,
        anomaly_direction,
        severity: severity.to_string(),
    }
}

fn percentile_rank(current: f64, values: &[f64]) -> f64 {
    if values.is_empty() {
        return 50.0;
    }
    let below_or_equal = values.iter().filter(|value| **value <= current).count();
    below_or_equal as f64 / values.len() as f64 * 100.0
}

fn safe_ratio(numerator: f64, denominator: f64) -> f64 {
    if denominator == 0.0 {
        0.0
    } else {
        numerator / denominator
    }
}

fn safe_percentage(numerator: f64, denominator: f64) -> f64 {
    safe_ratio(numerator, denominator) * 100.0
}

fn round_to(value: f64, decimals: i32) -> f64 {
    let factor = 10_f64.powi(decimals);
    (value * factor).round() / factor
}

fn parse_compare_ids(ids: &str) -> ApiResult<Vec<String>> {
    let mut parsed = Vec::new();
    for id in ids.split(',').map(str::trim).filter(|id| !id.is_empty()) {
        if has_sensitive_text(id) || id.len() > 80 {
            return Err(ApiError::BadRequest("invalid comparison id".to_string()));
        }
        if !parsed.iter().any(|existing| existing == id) {
            parsed.push(id.to_string());
        }
    }

    if !(2..=4).contains(&parsed.len()) {
        return Err(ApiError::BadRequest(
            "ids must include 2 to 4 unique values".to_string(),
        ));
    }

    Ok(parsed)
}

fn sort_by_requested_order<T, F>(mut items: Vec<T>, requested_ids: &[String], id_fn: F) -> Vec<T>
where
    F: Fn(&T) -> &str,
{
    items.sort_by_key(|item| {
        requested_ids
            .iter()
            .position(|id| id == id_fn(item))
            .unwrap_or(usize::MAX)
    });
    items
}

fn missing_ids<T, F>(requested_ids: &[String], items: &[T], id_fn: F) -> Vec<String>
where
    F: Fn(&T) -> &str,
{
    requested_ids
        .iter()
        .filter(|id| !items.iter().any(|item| id_fn(item) == id.as_str()))
        .cloned()
        .collect()
}

fn area_comparison_metrics(items: &[AreaScore]) -> Vec<ComparisonMetric> {
    vec![
        comparison_metric(
            "investment_score",
            "综合分",
            "分",
            "higher",
            items,
            |item| Some(item.investment_score),
        ),
        comparison_metric(
            "transaction_price_psm",
            "成交均价/㎡",
            "元/㎡",
            "neutral",
            items,
            |item| Some(item.transaction_price_psm),
        ),
        comparison_metric(
            "transaction_count",
            "成交套数",
            "套",
            "higher",
            items,
            |item| Some(item.transaction_count as f64),
        ),
        comparison_metric(
            "liquidity_score",
            "流动性",
            "分",
            "higher",
            items,
            |item| Some(item.liquidity_score),
        ),
        comparison_metric(
            "rent_support_score",
            "租金支撑",
            "分",
            "higher",
            items,
            |item| Some(item.rent_support_score),
        ),
        comparison_metric(
            "annual_rent_yield_pct",
            "租金收益率",
            "%",
            "higher",
            items,
            |item| Some(item.annual_rent_yield_pct),
        ),
        comparison_metric(
            "supply_risk_score",
            "供应风险",
            "分",
            "lower",
            items,
            |item| Some(item.supply_risk_score),
        ),
        comparison_metric(
            "valuation_pressure_score",
            "估值压力",
            "分",
            "lower",
            items,
            |item| Some(item.valuation_pressure_score),
        ),
    ]
}

fn neighborhood_comparison_metrics(items: &[NeighborhoodComparisonItem]) -> Vec<ComparisonMetric> {
    vec![
        comparison_metric(
            "investment_score",
            "综合分",
            "分",
            "higher",
            items,
            |item| Some(item.investment_score),
        ),
        comparison_metric(
            "transaction_price_psm",
            "成交均价/㎡",
            "元/㎡",
            "neutral",
            items,
            |item| Some(item.transaction_price_psm),
        ),
        comparison_metric(
            "transaction_count",
            "成交套数",
            "套",
            "higher",
            items,
            |item| Some(item.transaction_count as f64),
        ),
        comparison_metric(
            "liquidity_score",
            "流动性",
            "分",
            "higher",
            items,
            |item| Some(item.liquidity_score),
        ),
        comparison_metric(
            "rent_support_score",
            "租金支撑",
            "分",
            "higher",
            items,
            |item| Some(item.rent_support_score),
        ),
        comparison_metric(
            "building_age_score",
            "楼龄",
            "分",
            "higher",
            items,
            |item| Some(item.building_age_score),
        ),
        comparison_metric(
            "metro_distance_m",
            "地铁距离",
            "米",
            "lower",
            items,
            |item| item.metro_distance_m.map(|value| value as f64),
        ),
        comparison_metric(
            "available_units",
            "可售套数",
            "套",
            "neutral",
            items,
            |item| Some(item.available_units as f64),
        ),
    ]
}

fn comparison_metric<T, F>(
    key: &str,
    label: &str,
    unit: &str,
    direction: &str,
    items: &[T],
    value_fn: F,
) -> ComparisonMetric
where
    F: Fn(&T) -> Option<f64>,
    T: ComparisonId,
{
    ComparisonMetric {
        key: key.to_string(),
        label: label.to_string(),
        unit: unit.to_string(),
        direction: direction.to_string(),
        values: items
            .iter()
            .map(|item| ComparisonMetricValue {
                id: item.comparison_id().to_string(),
                value: value_fn(item),
            })
            .collect(),
    }
}

trait ComparisonId {
    fn comparison_id(&self) -> &str;
}

impl ComparisonId for AreaScore {
    fn comparison_id(&self) -> &str {
        &self.area_id
    }
}

impl ComparisonId for NeighborhoodComparisonItem {
    fn comparison_id(&self) -> &str {
        &self.neighborhood_id
    }
}

fn validate_month(month: &str) -> ApiResult<()> {
    let valid = month.len() == 7
        && month.as_bytes()[4] == b'-'
        && month[..4].chars().all(|item| item.is_ascii_digit())
        && month[5..].chars().all(|item| item.is_ascii_digit())
        && matches!(month[5..].parse::<u8>(), Ok(value) if (1..=12).contains(&value));

    if valid {
        Ok(())
    } else {
        Err(ApiError::BadRequest(
            "month must use YYYY-MM format".to_string(),
        ))
    }
}

fn validate_sha256(value: &str) -> ApiResult<()> {
    let valid = value.len() == 64
        && value
            .chars()
            .all(|item| item.is_ascii_hexdigit() && !item.is_ascii_uppercase());

    if valid {
        Ok(())
    } else {
        Err(ApiError::BadRequest(
            "sha256 must be a lowercase 64-character hex digest".to_string(),
        ))
    }
}

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 validate_report_type(value: &str) -> ApiResult<()> {
    if matches!(
        value,
        "monthly" | "weekly" | "area_special" | "neighborhood_memo"
    ) {
        Ok(())
    } else {
        Err(ApiError::BadRequest(
            "report_type must be monthly, weekly, area_special, or neighborhood_memo".to_string(),
        ))
    }
}

fn validate_report_status(value: &str) -> ApiResult<()> {
    if matches!(value, "draft" | "published" | "archived") {
        Ok(())
    } else {
        Err(ApiError::BadRequest(
            "status must be draft, published, or archived".to_string(),
        ))
    }
}

fn validate_generation_status(value: &str) -> ApiResult<()> {
    if matches!(value, "running" | "succeeded" | "failed") {
        Ok(())
    } else {
        Err(ApiError::BadRequest(
            "status must be running, succeeded, or failed".to_string(),
        ))
    }
}

fn has_sensitive_text(value: &str) -> bool {
    let lower = value.to_ascii_lowercase();
    lower.contains("password")
        || lower.contains("secret")
        || lower.contains("postgres://")
        || lower.contains("root_")
        || lower.contains("database_url")
        || value.starts_with("/Users/")
        || value.starts_with("/private/")
}

#[cfg(test)]
mod tests {
    use crate::models::{
        CreateDataSource, CreateIngestionRun, CreateRawArtifact, CreateReport,
        CreateReportGenerationRun, CreateScenarioRun, CreateWatchlistItem, FinishIngestionRun,
        ScenarioRun, SimilarNeighborhoodCandidate, UpdateWatchlistItem,
    };

    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() {
        assert!(validate_month("2026-05").is_ok());
    }

    #[test]
    fn rejects_invalid_month() {
        assert!(validate_month("2026-5").is_err());
        assert!(validate_month("2026-13").is_err());
        assert!(validate_month("abcd-05").is_err());
    }

    #[test]
    fn validates_watchlist_create_payload() {
        assert!(CreateWatchlistItem {
            neighborhood_id: None,
            area_id: None,
            target_price_psm: None,
            label: None,
            priority: None,
            trigger_operator: None,
            trigger_price_psm: None,
            notes: None,
        }
        .validate()
        .is_err());

        assert!(CreateWatchlistItem {
            neighborhood_id: None,
            area_id: Some("zhangjiang".to_string()),
            target_price_psm: Some(80_000.0),
            label: Some("核心".to_string()),
            priority: Some("high".to_string()),
            trigger_operator: Some("lte".to_string()),
            trigger_price_psm: Some(76_000.0),
            notes: None,
        }
        .validate()
        .is_ok());
    }

    #[test]
    fn validates_watchlist_update_payload() {
        assert!(UpdateWatchlistItem {
            target_price_psm: Some(-1.0),
            status: None,
            label: None,
            priority: None,
            trigger_operator: None,
            trigger_price_psm: None,
            notes: None,
            mark_reviewed: None,
        }
        .validate()
        .is_err());

        assert!(UpdateWatchlistItem {
            target_price_psm: None,
            status: Some("unknown".to_string()),
            label: None,
            priority: None,
            trigger_operator: None,
            trigger_price_psm: None,
            notes: None,
            mark_reviewed: None,
        }
        .validate()
        .is_err());

        assert!(UpdateWatchlistItem {
            target_price_psm: Some(80_000.0),
            status: Some("active".to_string()),
            label: Some("核心".to_string()),
            priority: Some("high".to_string()),
            trigger_operator: Some("lte".to_string()),
            trigger_price_psm: Some(76_000.0),
            notes: Some("跟踪产业兑现和挂牌变化".to_string()),
            mark_reviewed: Some(true),
        }
        .validate()
        .is_ok());
    }

    #[test]
    fn validates_data_source_payload() {
        assert!(CreateDataSource {
            name: "".to_string(),
            source_type: "official".to_string(),
            url: None,
            cadence: None,
            reliability: None,
            notes: None,
        }
        .validate()
        .is_err());

        assert!(CreateDataSource {
            name: "上海市统计局".to_string(),
            source_type: "official".to_string(),
            url: Some("https://tjj.sh.gov.cn/".to_string()),
            cadence: Some("monthly".to_string()),
            reliability: Some("high".to_string()),
            notes: None,
        }
        .validate()
        .is_ok());

        assert!(CreateDataSource {
            name: "unsafe".to_string(),
            source_type: "manual".to_string(),
            url: None,
            cadence: None,
            reliability: None,
            notes: Some("contains password marker".to_string()),
        }
        .validate()
        .is_err());
    }

    #[test]
    fn validates_ingestion_run_payloads() {
        assert!(CreateIngestionRun {
            source_id: Some(1),
            status: Some("running".to_string()),
            raw_uri: Some("raw/official/2026-05.csv".to_string()),
            raw_artifact_id: None,
            row_count: Some(10),
            error_message: None,
        }
        .validate()
        .is_ok());

        assert!(CreateIngestionRun {
            source_id: Some(1),
            status: Some("unknown".to_string()),
            raw_uri: None,
            raw_artifact_id: None,
            row_count: None,
            error_message: None,
        }
        .validate()
        .is_err());

        assert!(FinishIngestionRun {
            status: "succeeded".to_string(),
            row_count: Some(10),
            error_message: None,
        }
        .validate()
        .is_ok());

        assert!(FinishIngestionRun {
            status: "running".to_string(),
            row_count: None,
            error_message: None,
        }
        .validate()
        .is_err());
    }

    #[test]
    fn validates_raw_artifact_payload() {
        assert!(CreateRawArtifact {
            run_id: None,
            source_id: None,
            original_filename: "area.csv".to_string(),
            raw_uri: "raw://sha256/area.csv".to_string(),
            sha256: "a".repeat(64),
            size_bytes: 128,
            mime_type: Some("text/csv".to_string()),
            uploaded_by: Some("researcher".to_string()),
            notes: None,
        }
        .validate()
        .is_ok());

        assert!(CreateRawArtifact {
            run_id: None,
            source_id: None,
            original_filename: "area.csv".to_string(),
            raw_uri: "raw://sha256/area.csv".to_string(),
            sha256: "A".repeat(64),
            size_bytes: 128,
            mime_type: None,
            uploaded_by: None,
            notes: None,
        }
        .validate()
        .is_err());
    }

    #[test]
    fn validates_report_payload() {
        assert!(CreateReport {
            report_type: "monthly".to_string(),
            title: "2026-05 月报".to_string(),
            target_month: Some("2026-05".to_string()),
            area_id: None,
            neighborhood_id: None,
            summary: Some("核心结论".to_string()),
            content_markdown: "# 月报\n\n正文".to_string(),
            linked_metrics: Some(serde_json::json!({"month": "2026-05"})),
            status: Some("published".to_string()),
        }
        .validate()
        .is_ok());

        assert!(CreateReport {
            report_type: "unknown".to_string(),
            title: "bad".to_string(),
            target_month: None,
            area_id: None,
            neighborhood_id: None,
            summary: None,
            content_markdown: "# bad".to_string(),
            linked_metrics: None,
            status: None,
        }
        .validate()
        .is_err());

        assert!(CreateReport {
            report_type: "area_special".to_string(),
            title: "bad".to_string(),
            target_month: None,
            area_id: Some("zhangjiang".to_string()),
            neighborhood_id: None,
            summary: Some("contains password marker".to_string()),
            content_markdown: "# bad".to_string(),
            linked_metrics: None,
            status: None,
        }
        .validate()
        .is_err());
    }

    #[test]
    fn validates_report_generation_payload() {
        assert!(CreateReportGenerationRun {
            template_key: "monthly_market".to_string(),
            target_month: "2026-05".to_string(),
            parameters: Some(serde_json::json!({"scope": "all"})),
        }
        .validate()
        .is_ok());

        assert!(CreateReportGenerationRun {
            template_key: "".to_string(),
            target_month: "2026-05".to_string(),
            parameters: None,
        }
        .validate()
        .is_err());

        assert!(CreateReportGenerationRun {
            template_key: "monthly_market".to_string(),
            target_month: "2026-05".to_string(),
            parameters: Some(serde_json::json!(["bad"])),
        }
        .validate()
        .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());
        assert!(validate_sha256(&"G".repeat(64)).is_err());
        assert!(validate_sha256("short").is_err());
    }

    #[test]
    fn validates_compare_ids() {
        assert_eq!(
            parse_compare_ids("qiantan, zhangjiang,qiantan").unwrap(),
            vec!["qiantan".to_string(), "zhangjiang".to_string()]
        );
        assert!(parse_compare_ids("qiantan").is_err());
        assert!(parse_compare_ids("a,b,c,d,e").is_err());
        assert!(parse_compare_ids("a,password,b").is_err());
    }

    #[test]
    fn scores_similar_neighborhoods_with_explainable_price_delta() {
        let target =
            neighborhood_candidate("target", "zhangjiang", "浦东新区", 2020, 500, 89_000.0);
        let close_candidate =
            neighborhood_candidate("close", "zhangjiang", "浦东新区", 2018, 650, 86_000.0);
        let distant_candidate =
            neighborhood_candidate("distant", "hongqiao", "闵行区", 2002, 2400, 120_000.0);

        let close = similar_neighborhood(&target, close_candidate);
        let distant = similar_neighborhood(&target, distant_candidate);

        assert!(close.similarity_score > distant.similarity_score);
        assert_eq!(close.relative_price_pct, -3.4);
        assert!(close.reasons.contains(&"same_area".to_string()));
        assert!(close
            .reasons
            .iter()
            .any(|reason| reason.starts_with("price_gap_pct:")));
    }

    fn neighborhood_candidate(
        neighborhood_id: &str,
        area_id: &str,
        district: &str,
        built_year: i32,
        metro_distance_m: i32,
        transaction_price_psm: f64,
    ) -> SimilarNeighborhoodCandidate {
        SimilarNeighborhoodCandidate {
            neighborhood_id: neighborhood_id.to_string(),
            area_id: area_id.to_string(),
            name: neighborhood_id.to_string(),
            area_name: area_id.to_string(),
            district: district.to_string(),
            built_year: Some(built_year),
            property_type: "商品住宅".to_string(),
            metro_distance_m: Some(metro_distance_m),
            school_quality: Some("normal".to_string()),
            month: "2026-05".to_string(),
            investment_score: 70.0,
            recommendation: "观察池".to_string(),
            transaction_price_psm,
            annual_rent_yield_pct: 2.0,
            liquidity_score: 80.0,
            location_score: 75.0,
            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);
    }
}