feat: add area score model runs
This commit is contained in:
@@ -5,13 +5,15 @@ use serde::Serialize;
|
||||
use crate::error::{ApiError, ApiResult};
|
||||
use crate::import_templates::{execute_import_payload, validate_import_payload};
|
||||
use crate::models::{
|
||||
AreaScore, AreaScoreSummary, CreateDataSource, CreateIngestionRun, CreateRawArtifact,
|
||||
CreateWatchlistItem, DataSource, FinishIngestionRun, ImportExecutionRequest,
|
||||
AreaScore, AreaScoreLineage, AreaScoreLineageQuery, AreaScoreModelRunResponse,
|
||||
AreaScoreSummary, CreateAreaScoreModelRun, CreateDataSource, CreateIngestionRun,
|
||||
CreateRawArtifact, CreateWatchlistItem, DataSource, FinishIngestionRun, ImportExecutionRequest,
|
||||
ImportExecutionResponse, ImportValidationRequest, ImportValidationResponse, IngestionRun,
|
||||
IngestionRunQuery, MarketOverview, MonthQuery, Neighborhood, NeighborhoodQuery,
|
||||
IngestionRunQuery, MarketOverview, ModelRun, MonthQuery, Neighborhood, NeighborhoodQuery,
|
||||
NeighborhoodScore, NeighborhoodScoreQuery, RawArtifact, RawArtifactQuery, UpdateWatchlistItem,
|
||||
WatchlistItem,
|
||||
};
|
||||
use crate::scoring::{compute_area_scores, previous_month, AreaMetricRow, ComputedAreaScore};
|
||||
use crate::state::AppState;
|
||||
|
||||
#[derive(Debug, Serialize)]
|
||||
@@ -130,6 +132,143 @@ pub async fn market_overview(
|
||||
}))
|
||||
}
|
||||
|
||||
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 = 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(query.month)
|
||||
.fetch_optional(&state.pool)
|
||||
.await?
|
||||
.ok_or_else(|| ApiError::BadRequest("area score lineage not found".to_string()))?;
|
||||
|
||||
Ok(Json(lineage))
|
||||
}
|
||||
|
||||
pub async fn list_neighborhoods(
|
||||
State(state): State<AppState>,
|
||||
Query(query): Query<NeighborhoodQuery>,
|
||||
@@ -739,6 +878,234 @@ async fn fetch_area_scores(state: &AppState, month: &str) -> Result<Vec<AreaScor
|
||||
.await
|
||||
}
|
||||
|
||||
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(¤t_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?;
|
||||
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
|
||||
}
|
||||
|
||||
fn average(values: impl Iterator<Item = f64>) -> f64 {
|
||||
let (sum, count) = values.fold((0.0, 0usize), |(sum, count), value| {
|
||||
(sum + value, count + 1)
|
||||
@@ -781,6 +1148,17 @@ fn validate_sha256(value: &str) -> ApiResult<()> {
|
||||
}
|
||||
}
|
||||
|
||||
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::{
|
||||
|
||||
Reference in New Issue
Block a user