diff --git a/apps/api/src/handlers.rs b/apps/api/src/handlers.rs
index 9eee70f..ef14fe2 100644
--- a/apps/api/src/handlers.rs
+++ b/apps/api/src/handlers.rs
@@ -3,11 +3,12 @@ use axum::Json;
 use serde::Serialize;
 
 use crate::error::{ApiError, ApiResult};
+use crate::import_templates::validate_import_payload;
 use crate::models::{
     AreaScore, AreaScoreSummary, CreateDataSource, CreateIngestionRun, CreateWatchlistItem,
-    DataSource, FinishIngestionRun, IngestionRun, IngestionRunQuery, MarketOverview, MonthQuery,
-    Neighborhood, NeighborhoodQuery, NeighborhoodScore, NeighborhoodScoreQuery,
-    UpdateWatchlistItem, WatchlistItem,
+    DataSource, FinishIngestionRun, ImportValidationRequest, ImportValidationResponse,
+    IngestionRun, IngestionRunQuery, MarketOverview, MonthQuery, Neighborhood, NeighborhoodQuery,
+    NeighborhoodScore, NeighborhoodScoreQuery, UpdateWatchlistItem, WatchlistItem,
 };
 use crate::state::AppState;
 
@@ -32,6 +33,13 @@ pub async fn ready(State(state): State<AppState>) -> ApiResult<Json<HealthRespon
     }))
 }
 
+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 list_area_scores(
     State(state): State<AppState>,
     Query(query): Query<MonthQuery>,
diff --git a/apps/api/src/import_templates.rs b/apps/api/src/import_templates.rs
new file mode 100644
index 0000000..81dcc73
--- /dev/null
+++ b/apps/api/src/import_templates.rs
@@ -0,0 +1,389 @@
+use std::collections::{HashMap, HashSet};
+
+use serde::Deserialize;
+use serde_json::Value;
+
+use crate::models::{ImportValidationError, ImportValidationRequest, ImportValidationResponse};
+
+const IMPORT_TEMPLATE_SPEC: &str = include_str!("../../../config/import_templates.json");
+
+#[derive(Debug, Deserialize)]
+struct TemplateSpec {
+    columns: Vec<ColumnSpec>,
+}
+
+#[derive(Debug, Deserialize)]
+struct ColumnSpec {
+    name: String,
+    #[serde(rename = "type")]
+    column_type: String,
+    required: bool,
+    min: Option<f64>,
+    max: Option<f64>,
+    allowed_values: Option<Vec<String>>,
+}
+
+pub fn validate_import_payload(
+    payload: &ImportValidationRequest,
+) -> Result<ImportValidationResponse, String> {
+    let specs = load_specs()?;
+    let spec = specs.get(&payload.template_name).ok_or_else(|| {
+        let mut valid_names = specs.keys().cloned().collect::<Vec<_>>();
+        valid_names.sort();
+        format!(
+            "unknown template '{}', expected one of: {}",
+            payload.template_name,
+            valid_names.join(", ")
+        )
+    })?;
+
+    let mut errors = Vec::new();
+    let expected_columns = spec
+        .columns
+        .iter()
+        .map(|column| column.name.as_str())
+        .collect::<HashSet<_>>();
+    let seen_columns = payload
+        .rows
+        .iter()
+        .flat_map(|row| row.keys().map(String::as_str))
+        .collect::<HashSet<_>>();
+
+    for column in &spec.columns {
+        if !seen_columns.contains(column.name.as_str()) {
+            errors.push(ImportValidationError::new(
+                0,
+                &column.name,
+                "missing required header",
+            ));
+        }
+    }
+
+    let mut unexpected_columns = seen_columns
+        .iter()
+        .copied()
+        .filter(|column| !expected_columns.contains(column))
+        .collect::<Vec<_>>();
+    unexpected_columns.sort_unstable();
+    for column in unexpected_columns {
+        errors.push(ImportValidationError::new(0, column, "unexpected header"));
+    }
+
+    if payload.rows.is_empty() {
+        errors.push(ImportValidationError::new(0, "*", "file has no data rows"));
+    }
+
+    for (index, row) in payload.rows.iter().enumerate() {
+        let row_number = index + 2;
+        for column in &spec.columns {
+            let value = normalize_value(row.get(&column.name));
+            if column.required && value.is_empty() {
+                errors.push(ImportValidationError::new(
+                    row_number,
+                    &column.name,
+                    "value is required",
+                ));
+                continue;
+            }
+            if value.is_empty() {
+                continue;
+            }
+            errors.extend(validate_value(row_number, &column.name, &value, column));
+        }
+    }
+
+    Ok(ImportValidationResponse {
+        template_name: payload.template_name.clone(),
+        row_count: payload.rows.len(),
+        valid: errors.is_empty(),
+        errors,
+    })
+}
+
+fn load_specs() -> Result<HashMap<String, TemplateSpec>, String> {
+    serde_json::from_str(IMPORT_TEMPLATE_SPEC)
+        .map_err(|error| format!("failed to parse import template specs: {error}"))
+}
+
+fn normalize_value(value: Option<&Value>) -> String {
+    match value {
+        None | Some(Value::Null) => String::new(),
+        Some(Value::String(value)) => value.trim().to_string(),
+        Some(Value::Number(value)) => value.to_string(),
+        Some(Value::Bool(value)) => value.to_string(),
+        Some(value) => value.to_string(),
+    }
+}
+
+fn validate_value(
+    row_number: usize,
+    column_name: &str,
+    value: &str,
+    column: &ColumnSpec,
+) -> Vec<ImportValidationError> {
+    if has_sensitive_text(value) {
+        return vec![ImportValidationError::new(
+            row_number,
+            column_name,
+            "value appears to contain credentials or a sensitive local path",
+        )];
+    }
+
+    match column.column_type.as_str() {
+        "text" => Vec::new(),
+        "month" => {
+            if is_valid_month(value) {
+                Vec::new()
+            } else {
+                vec![ImportValidationError::new(
+                    row_number,
+                    column_name,
+                    "must be YYYY-MM",
+                )]
+            }
+        }
+        "date" => {
+            if is_valid_date(value) {
+                Vec::new()
+            } else {
+                vec![ImportValidationError::new(
+                    row_number,
+                    column_name,
+                    "must be YYYY-MM-DD",
+                )]
+            }
+        }
+        "url" => {
+            if is_valid_url(value) {
+                Vec::new()
+            } else {
+                vec![ImportValidationError::new(
+                    row_number,
+                    column_name,
+                    "must be http(s) URL",
+                )]
+            }
+        }
+        "enum" => {
+            let allowed_values = column.allowed_values.as_deref().unwrap_or_default();
+            if allowed_values.iter().any(|allowed| allowed == value) {
+                Vec::new()
+            } else {
+                vec![ImportValidationError::new(
+                    row_number,
+                    column_name,
+                    format!("must be one of: {}", allowed_values.join(", ")),
+                )]
+            }
+        }
+        "integer" => match parse_integer(value) {
+            Some(parsed) => validate_numeric_range(row_number, column_name, parsed as f64, column),
+            _ => vec![ImportValidationError::new(
+                row_number,
+                column_name,
+                "must be an integer",
+            )],
+        },
+        "decimal" => match value.parse::<f64>() {
+            Ok(parsed) if parsed.is_finite() => {
+                validate_numeric_range(row_number, column_name, parsed, column)
+            }
+            _ => vec![ImportValidationError::new(
+                row_number,
+                column_name,
+                "must be a number",
+            )],
+        },
+        unknown => vec![ImportValidationError::new(
+            row_number,
+            column_name,
+            format!("unknown type '{unknown}'"),
+        )],
+    }
+}
+
+fn validate_numeric_range(
+    row_number: usize,
+    column_name: &str,
+    value: f64,
+    column: &ColumnSpec,
+) -> Vec<ImportValidationError> {
+    let mut errors = Vec::new();
+    if let Some(minimum) = column.min {
+        if value < minimum {
+            errors.push(ImportValidationError::new(
+                row_number,
+                column_name,
+                format!("must be >= {}", format_number(minimum)),
+            ));
+        }
+    }
+    if let Some(maximum) = column.max {
+        if value > maximum {
+            errors.push(ImportValidationError::new(
+                row_number,
+                column_name,
+                format!("must be <= {}", format_number(maximum)),
+            ));
+        }
+    }
+    errors
+}
+
+fn format_number(value: f64) -> String {
+    if value.fract() == 0.0 {
+        format!("{}", value as i64)
+    } else {
+        value.to_string()
+    }
+}
+
+fn parse_integer(value: &str) -> Option<i64> {
+    let digits = value.strip_prefix('-').unwrap_or(value);
+    if digits.is_empty() || !digits.chars().all(|item| item.is_ascii_digit()) {
+        return None;
+    }
+    value.parse::<i64>().ok()
+}
+
+fn is_valid_month(value: &str) -> bool {
+    if value.len() != 7 || value.as_bytes()[4] != b'-' {
+        return false;
+    }
+    value[..4].chars().all(|item| item.is_ascii_digit())
+        && value[5..].chars().all(|item| item.is_ascii_digit())
+        && matches!(value[5..].parse::<u8>(), Ok(month) if (1..=12).contains(&month))
+}
+
+fn is_valid_date(value: &str) -> bool {
+    if value.len() != 10 || value.as_bytes()[4] != b'-' || value.as_bytes()[7] != b'-' {
+        return false;
+    }
+    if !value
+        .chars()
+        .enumerate()
+        .all(|(index, item)| index == 4 || index == 7 || item.is_ascii_digit())
+    {
+        return false;
+    }
+
+    let year = value[..4].parse::<i32>().unwrap_or_default();
+    let month = value[5..7].parse::<u8>().unwrap_or_default();
+    let day = value[8..].parse::<u8>().unwrap_or_default();
+    if !(1..=12).contains(&month) {
+        return false;
+    }
+    let max_day = match month {
+        1 | 3 | 5 | 7 | 8 | 10 | 12 => 31,
+        4 | 6 | 9 | 11 => 30,
+        2 if is_leap_year(year) => 29,
+        2 => 28,
+        _ => 0,
+    };
+    (1..=max_day).contains(&day)
+}
+
+fn is_leap_year(year: i32) -> bool {
+    (year % 4 == 0 && year % 100 != 0) || year % 400 == 0
+}
+
+fn is_valid_url(value: &str) -> bool {
+    (value.starts_with("http://") || value.starts_with("https://"))
+        && value
+            .split_once("://")
+            .is_some_and(|(_, rest)| rest.contains('.') && !rest.starts_with('.'))
+}
+
+fn has_sensitive_text(value: &str) -> bool {
+    let lower = value.to_ascii_lowercase();
+    lower.contains("password")
+        || lower.contains("secret")
+        || lower.contains("postgres://")
+        || lower.contains("database_url")
+        || lower.contains("root_")
+        || value.starts_with("/Users/")
+        || value.starts_with("/private/")
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use serde_json::json;
+
+    use crate::models::ImportValidationRequest;
+
+    use super::{is_valid_date, validate_import_payload};
+
+    #[test]
+    fn validates_good_import_rows() {
+        let request = ImportValidationRequest {
+            template_name: "area_monthly_metrics".to_string(),
+            rows: vec![HashMap::from([
+                ("area_id".to_string(), json!("qiantan")),
+                ("month".to_string(), json!("2026-05")),
+                ("transaction_count".to_string(), json!("93")),
+                ("transaction_price_psm".to_string(), json!("122500")),
+                ("listing_count".to_string(), json!("380")),
+                ("listing_price_psm".to_string(), json!("126500")),
+                ("median_days_on_market".to_string(), json!("53")),
+                ("rent_price_psm".to_string(), json!("192")),
+                ("new_supply_units".to_string(), json!("60")),
+                ("land_residential_gfa_sqm".to_string(), json!("0")),
+                ("mortgage_rate_pct".to_string(), json!("3.35")),
+                ("policy_signal".to_string(), json!("1")),
+                ("source".to_string(), json!("manual_research")),
+            ])],
+        };
+
+        let result = validate_import_payload(&request).expect("valid spec");
+
+        assert!(result.valid);
+        assert_eq!(result.row_count, 1);
+        assert!(result.errors.is_empty());
+    }
+
+    #[test]
+    fn reports_bad_import_rows() {
+        let request = ImportValidationRequest {
+            template_name: "area_monthly_metrics".to_string(),
+            rows: vec![HashMap::from([
+                ("area_id".to_string(), json!("")),
+                ("month".to_string(), json!("2026-13")),
+                ("transaction_count".to_string(), json!("-1")),
+                ("transaction_price_psm".to_string(), json!("abc")),
+                ("listing_count".to_string(), json!("10")),
+                ("listing_price_psm".to_string(), json!("100")),
+                ("median_days_on_market".to_string(), json!("30")),
+                ("rent_price_psm".to_string(), json!("3")),
+                ("new_supply_units".to_string(), json!("0")),
+                ("land_residential_gfa_sqm".to_string(), json!("0")),
+                ("mortgage_rate_pct".to_string(), json!("3.35")),
+                ("policy_signal".to_string(), json!("3")),
+                ("source".to_string(), json!("manual")),
+            ])],
+        };
+
+        let result = validate_import_payload(&request).expect("valid spec");
+        let messages = result
+            .errors
+            .iter()
+            .map(|error| (error.column.as_str(), error.message.as_str()))
+            .collect::<Vec<_>>();
+
+        assert!(!result.valid);
+        assert!(messages.contains(&("area_id", "value is required")));
+        assert!(messages.contains(&("month", "must be YYYY-MM")));
+        assert!(messages.contains(&("transaction_count", "must be >= 0")));
+        assert!(messages.contains(&("transaction_price_psm", "must be a number")));
+        assert!(messages.contains(&("policy_signal", "must be <= 2")));
+    }
+
+    #[test]
+    fn validates_real_calendar_dates() {
+        assert!(is_valid_date("2026-02-28"));
+        assert!(!is_valid_date("2026-02-29"));
+        assert!(is_valid_date("2024-02-29"));
+        assert!(!is_valid_date("2026-13-01"));
+    }
+}
diff --git a/apps/api/src/main.rs b/apps/api/src/main.rs
index 597bf0a..2352223 100644
--- a/apps/api/src/main.rs
+++ b/apps/api/src/main.rs
@@ -2,6 +2,7 @@ mod config;
 mod db;
 mod error;
 mod handlers;
+mod import_templates;
 mod models;
 mod routes;
 mod state;
diff --git a/apps/api/src/models.rs b/apps/api/src/models.rs
index 544ae29..f0a1897 100644
--- a/apps/api/src/models.rs
+++ b/apps/api/src/models.rs
@@ -1,4 +1,7 @@
+use std::collections::HashMap;
+
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 use sqlx::FromRow;
 
 #[derive(Debug, Deserialize)]
@@ -22,6 +25,37 @@ pub struct IngestionRunQuery {
     pub source_id: Option<i64>,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct ImportValidationRequest {
+    pub template_name: String,
+    pub rows: Vec<HashMap<String, Value>>,
+}
+
+#[derive(Debug, Clone, Serialize, PartialEq)]
+pub struct ImportValidationError {
+    pub row: usize,
+    pub column: String,
+    pub message: String,
+}
+
+impl ImportValidationError {
+    pub fn new(row: usize, column: impl Into<String>, message: impl Into<String>) -> Self {
+        Self {
+            row,
+            column: column.into(),
+            message: message.into(),
+        }
+    }
+}
+
+#[derive(Debug, Serialize)]
+pub struct ImportValidationResponse {
+    pub template_name: String,
+    pub row_count: usize,
+    pub valid: bool,
+    pub errors: Vec<ImportValidationError>,
+}
+
 #[derive(Debug, Clone, Serialize, FromRow)]
 pub struct AreaScore {
     pub area_id: String,
diff --git a/apps/api/src/routes.rs b/apps/api/src/routes.rs
index 1b1338c..5c811ad 100644
--- a/apps/api/src/routes.rs
+++ b/apps/api/src/routes.rs
@@ -7,7 +7,7 @@ use crate::handlers::{
     archive_watchlist_item, create_data_source, create_ingestion_run, create_watchlist_item,
     finish_ingestion_run, get_neighborhood, health, list_area_scores, list_data_sources,
     list_ingestion_runs, list_neighborhood_scores, list_neighborhoods, list_watchlist_items,
-    market_overview, ready, update_watchlist_item,
+    market_overview, ready, update_watchlist_item, validate_import,
 };
 use crate::state::AppState;
 
@@ -31,6 +31,7 @@ pub fn build_router(state: AppState) -> Router {
                     "/ingestion-runs/{run_id}/finish",
                     patch(finish_ingestion_run),
                 )
+                .route("/imports/validate", axum::routing::post(validate_import))
                 .route("/market/overview", get(market_overview))
                 .route("/neighborhoods", get(list_neighborhoods))
                 .route("/neighborhoods/scores", get(list_neighborhood_scores))
diff --git a/config/import_templates.json b/config/import_templates.json
new file mode 100644
index 0000000..a9c3867
--- /dev/null
+++ b/config/import_templates.json
@@ -0,0 +1,295 @@
+{
+  "area_monthly_metrics": {
+    "description": "Monthly area-level housing market metrics.",
+    "target_table": "silver.area_monthly_metrics",
+    "primary_key": ["area_id", "month"],
+    "columns": [
+      {
+        "name": "area_id",
+        "type": "text",
+        "required": true,
+        "description": "Stable area identifier from silver.areas."
+      },
+      {
+        "name": "month",
+        "type": "month",
+        "required": true,
+        "description": "Metric month in YYYY-MM format."
+      },
+      {
+        "name": "transaction_count",
+        "type": "integer",
+        "required": true,
+        "min": 0,
+        "description": "Observed second-hand transaction count."
+      },
+      {
+        "name": "transaction_price_psm",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Average transaction price, RMB per square meter."
+      },
+      {
+        "name": "listing_count",
+        "type": "integer",
+        "required": true,
+        "min": 0,
+        "description": "Active listing count."
+      },
+      {
+        "name": "listing_price_psm",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Average listing price, RMB per square meter."
+      },
+      {
+        "name": "median_days_on_market",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Median days on market."
+      },
+      {
+        "name": "rent_price_psm",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Monthly rent, RMB per square meter."
+      },
+      {
+        "name": "new_supply_units",
+        "type": "integer",
+        "required": true,
+        "min": 0,
+        "description": "New residential supply units."
+      },
+      {
+        "name": "land_residential_gfa_sqm",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Residential land transaction GFA, square meters."
+      },
+      {
+        "name": "mortgage_rate_pct",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "max": 20,
+        "description": "Mainstream mortgage rate percentage."
+      },
+      {
+        "name": "policy_signal",
+        "type": "integer",
+        "required": true,
+        "min": -2,
+        "max": 2,
+        "description": "Policy direction signal from -2 to 2."
+      },
+      {
+        "name": "source",
+        "type": "text",
+        "required": true,
+        "description": "Human-readable source name or source code."
+      }
+    ]
+  },
+  "neighborhood_monthly_metrics": {
+    "description": "Monthly neighborhood-level housing market metrics.",
+    "target_table": "silver.neighborhood_monthly_metrics",
+    "primary_key": ["neighborhood_id", "month"],
+    "columns": [
+      {
+        "name": "neighborhood_id",
+        "type": "text",
+        "required": true,
+        "description": "Stable neighborhood identifier from silver.neighborhoods."
+      },
+      {
+        "name": "month",
+        "type": "month",
+        "required": true,
+        "description": "Metric month in YYYY-MM format."
+      },
+      {
+        "name": "transaction_count",
+        "type": "integer",
+        "required": true,
+        "min": 0,
+        "description": "Observed transaction count."
+      },
+      {
+        "name": "transaction_price_psm",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Average transaction price, RMB per square meter."
+      },
+      {
+        "name": "listing_count",
+        "type": "integer",
+        "required": true,
+        "min": 0,
+        "description": "Active listing count."
+      },
+      {
+        "name": "listing_price_psm",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Average listing price, RMB per square meter."
+      },
+      {
+        "name": "rent_price_psm",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Monthly rent, RMB per square meter."
+      },
+      {
+        "name": "median_days_on_market",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Median days on market."
+      },
+      {
+        "name": "available_units",
+        "type": "integer",
+        "required": true,
+        "min": 0,
+        "description": "Currently available units in the neighborhood sample."
+      },
+      {
+        "name": "source",
+        "type": "text",
+        "required": true,
+        "description": "Human-readable source name or source code."
+      }
+    ]
+  },
+  "policy_events": {
+    "description": "Policy events that can affect market liquidity, credit or demand.",
+    "target_table": "raw.policy_events",
+    "primary_key": ["event_date", "title"],
+    "columns": [
+      {
+        "name": "event_date",
+        "type": "date",
+        "required": true,
+        "description": "Policy event date in YYYY-MM-DD format."
+      },
+      {
+        "name": "level",
+        "type": "enum",
+        "required": true,
+        "allowed_values": ["national", "shanghai", "district"],
+        "description": "Policy jurisdiction level."
+      },
+      {
+        "name": "title",
+        "type": "text",
+        "required": true,
+        "description": "Short policy title."
+      },
+      {
+        "name": "impact_direction",
+        "type": "integer",
+        "required": true,
+        "min": -2,
+        "max": 2,
+        "description": "Expected impact from -2 tightening to 2 easing."
+      },
+      {
+        "name": "notes",
+        "type": "text",
+        "required": false,
+        "description": "Analyst notes and interpretation."
+      },
+      {
+        "name": "source_url",
+        "type": "url",
+        "required": false,
+        "description": "Official notice or source URL."
+      },
+      {
+        "name": "source",
+        "type": "text",
+        "required": true,
+        "description": "Human-readable source name or source code."
+      }
+    ]
+  },
+  "land_sales": {
+    "description": "Residential land transaction records used for supply pressure analysis.",
+    "target_table": "raw.land_sales",
+    "primary_key": ["parcel_id"],
+    "columns": [
+      {
+        "name": "parcel_id",
+        "type": "text",
+        "required": true,
+        "description": "Stable land parcel identifier from the source."
+      },
+      {
+        "name": "area_id",
+        "type": "text",
+        "required": true,
+        "description": "Mapped area identifier from silver.areas."
+      },
+      {
+        "name": "transaction_date",
+        "type": "date",
+        "required": true,
+        "description": "Land transaction date in YYYY-MM-DD format."
+      },
+      {
+        "name": "parcel_name",
+        "type": "text",
+        "required": true,
+        "description": "Land parcel name."
+      },
+      {
+        "name": "residential_gfa_sqm",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Residential planned gross floor area, square meters."
+      },
+      {
+        "name": "transaction_price_total_cny",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Total transaction price in RMB."
+      },
+      {
+        "name": "floor_price_cny_psm",
+        "type": "decimal",
+        "required": true,
+        "min": 0,
+        "description": "Implied floor price, RMB per square meter."
+      },
+      {
+        "name": "buyer",
+        "type": "text",
+        "required": false,
+        "description": "Winning bidder or buyer."
+      },
+      {
+        "name": "source_url",
+        "type": "url",
+        "required": false,
+        "description": "Announcement or source URL."
+      },
+      {
+        "name": "source",
+        "type": "text",
+        "required": true,
+        "description": "Human-readable source name or source code."
+      }
+    ]
+  }
+}
diff --git a/docs/import_templates/README.md b/docs/import_templates/README.md
new file mode 100644
index 0000000..fd67ba7
--- /dev/null
+++ b/docs/import_templates/README.md
@@ -0,0 +1,74 @@
+# CSV/Excel 导入模板
+
+本目录定义真实数据进入系统前的标准格式。模板当前以 CSV 提供，Excel 可按相同列名保存为首行表头后再导出为 CSV。
+
+## 使用原则
+
+- 表头必须与模板完全一致。
+- 文件编码建议使用 UTF-8 或 UTF-8 with BOM。
+- 所有金额和面积字段只填写数字，不带单位、逗号或中文说明。
+- `source` 必须填写可识别的数据来源名称，不允许写数据库连接串、密码、本地敏感路径。
+- 单次导入必须先创建或关联一条 `audit.ingestion_runs` 记录。
+- 校验通过不代表数据一定正确，只代表结构、类型和值域满足入库前置条件。
+
+## 模板清单
+
+| 模板 | 目标表 | 主键 | 用途 |
+| --- | --- | --- | --- |
+| `area_monthly_metrics.csv` | `silver.area_monthly_metrics` | `area_id`, `month` | 板块月度成交、挂牌、租金、供应、信贷和政策信号 |
+| `neighborhood_monthly_metrics.csv` | `silver.neighborhood_monthly_metrics` | `neighborhood_id`, `month` | 小区月度成交、挂牌、租金和可售样本 |
+| `policy_events.csv` | `raw.policy_events` | `event_date`, `title` | 政策事件留痕和人工影响评分 |
+| `land_sales.csv` | `raw.land_sales` | `parcel_id` | 土地成交记录，用于供给压力研究 |
+
+## 字段规则
+
+字段规格的机器可读版本位于 `config/import_templates.json`。Python CLI 和 Rust API 均应以该文件为准。
+
+### 通用类型
+
+| 类型 | 规则 |
+| --- | --- |
+| `text` | 非空文本，若字段非必填则可为空 |
+| `integer` | 整数，可配置最小值和最大值 |
+| `decimal` | 数字，可配置最小值和最大值 |
+| `month` | `YYYY-MM`，月份必须是 01-12 |
+| `date` | `YYYY-MM-DD`，日期必须真实存在 |
+| `url` | 空值或 `http://` / `https://` URL |
+| `enum` | 必须属于规格中的 `allowed_values` |
+
+## CLI 校验
+
+```bash
+PYTHONPATH=src python3 -m shanghai_housing validate-import \
+  --template area_monthly_metrics \
+  --file docs/import_templates/area_monthly_metrics.csv
+```
+
+通过时输出行数和模板名；失败时逐行返回错误信息。
+
+## API 校验
+
+```bash
+curl -X POST http://127.0.0.1:8080/api/v1/imports/validate \
+  -H 'content-type: application/json' \
+  -d '{
+    "template_name": "area_monthly_metrics",
+    "rows": [
+      {
+        "area_id": "qiantan",
+        "month": "2026-05",
+        "transaction_count": "93",
+        "transaction_price_psm": "122500",
+        "listing_count": "380",
+        "listing_price_psm": "126500",
+        "median_days_on_market": "53",
+        "rent_price_psm": "192",
+        "new_supply_units": "60",
+        "land_residential_gfa_sqm": "0",
+        "mortgage_rate_pct": "3.35",
+        "policy_signal": "1",
+        "source": "manual_research"
+      }
+    ]
+  }'
+```
diff --git a/docs/import_templates/area_monthly_metrics.csv b/docs/import_templates/area_monthly_metrics.csv
new file mode 100644
index 0000000..c7a3aa3
--- /dev/null
+++ b/docs/import_templates/area_monthly_metrics.csv
@@ -0,0 +1,2 @@
+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
+qiantan,2026-05,93,122500,380,126500,53,192,60,0,3.35,1,manual_research
diff --git a/docs/import_templates/land_sales.csv b/docs/import_templates/land_sales.csv
new file mode 100644
index 0000000..a6ebb38
--- /dev/null
+++ b/docs/import_templates/land_sales.csv
@@ -0,0 +1,2 @@
+parcel_id,area_id,transaction_date,parcel_name,residential_gfa_sqm,transaction_price_total_cny,floor_price_cny_psm,buyer,source_url,source
+sh-2026-001,zhangjiang,2026-05-28,张江示例住宅地块,36000,1800000000,50000,示例房企,https://example.com/land-sale,official_auction
diff --git a/docs/import_templates/neighborhood_monthly_metrics.csv b/docs/import_templates/neighborhood_monthly_metrics.csv
new file mode 100644
index 0000000..461d2ce
--- /dev/null
+++ b/docs/import_templates/neighborhood_monthly_metrics.csv
@@ -0,0 +1,2 @@
+neighborhood_id,month,transaction_count,transaction_price_psm,listing_count,listing_price_psm,rent_price_psm,median_days_on_market,available_units,source
+qiantan_a,2026-05,12,124000,38,128000,195,50,38,manual_research
diff --git a/docs/import_templates/policy_events.csv b/docs/import_templates/policy_events.csv
new file mode 100644
index 0000000..8001127
--- /dev/null
+++ b/docs/import_templates/policy_events.csv
@@ -0,0 +1,2 @@
+event_date,level,title,impact_direction,notes,source_url,source
+2026-05-17,shanghai,优化住房信贷政策,1,降低置换成本,https://example.com/policy,official_notice
diff --git a/docs/product_roadmap.md b/docs/product_roadmap.md
index fe23914..19d5255 100644
--- a/docs/product_roadmap.md
+++ b/docs/product_roadmap.md
@@ -77,6 +77,8 @@
 
 ### M1.3 CSV/Excel 导入模板
 
+状态：已完成。
+
 目标：
 
 - 建立真实数据的标准导入模板。
@@ -87,23 +89,27 @@
 - `docs/import_templates/`
 - 后端导入校验接口。
 - Python analytics 导入脚本。
+- `config/import_templates.json` 作为模板规格源。
 
 验收标准：
 
-- 模板字段和数据库字段一一对应。
+- 模板字段和现有研究表字段一一对应。
 - 错误数据能返回明确错误信息。
-- 导入结果写入 `raw`/`silver`/`audit`。
+- Python CLI 可以校验 CSV/Excel，并将已支持模板写入本地研究库。
+- API 可以校验导入行的表头、类型、值域、日期、枚举和敏感文本。
 
 ### M1.4 原始数据留存与文件哈希
 
 目标：
 
-- 每次导入都记录原始文件位置、哈希、大小和导入人。
+- 每次导入都记录原始文件位置、哈希、大小、导入人和入库批次。
+- 正式导入结果写入 `raw`/`silver`/`audit`，并与 `audit.ingestion_runs` 关联。
 
 交付物：
 
 - `audit.raw_artifacts`
 - 导入批次关联 raw artifact。
+- PostgreSQL 导入执行接口。
 
 验收标准：
 
diff --git a/pyproject.toml b/pyproject.toml
index b46e988..81418ef 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -3,7 +3,9 @@ name = "shanghai-housing-research"
 version = "0.1.0"
 description = "A lightweight Shanghai housing market research and investment analysis system."
 requires-python = ">=3.9"
-dependencies = []
+dependencies = [
+    "openpyxl>=3.1",
+]
 
 [tool.setuptools.packages.find]
 where = ["src"]
diff --git a/src/shanghai_housing/cli.py b/src/shanghai_housing/cli.py
index 1523135..88938f6 100644
--- a/src/shanghai_housing/cli.py
+++ b/src/shanghai_housing/cli.py
@@ -4,8 +4,9 @@ import argparse
 from pathlib import Path
 from typing import Optional
 
-from .db import DEFAULT_DB_PATH, connect, init_db, load_sample_data
+from .db import DEFAULT_DB_PATH, connect, import_template_file, init_db, load_sample_data
 from .indicators import compute_area_scores
+from .import_templates import ImportTemplateError, validate_tabular_file
 from .reporting import format_scores_table, write_monthly_report
 
 
@@ -36,6 +37,38 @@ def build_parser() -> argparse.ArgumentParser:
         help="Output Markdown path. Default: reports/monthly-YYYY-MM.md",
     )
 
+    validate_import_parser = subparsers.add_parser(
+        "validate-import",
+        help="Validate a CSV or Excel import file against a named template.",
+    )
+    validate_import_parser.add_argument(
+        "--template",
+        required=True,
+        help="Template name, for example area_monthly_metrics.",
+    )
+    validate_import_parser.add_argument(
+        "--file",
+        type=Path,
+        required=True,
+        help="CSV, XLSX, or XLSM file to validate.",
+    )
+
+    import_parser = subparsers.add_parser(
+        "import-file",
+        help="Validate and import a CSV or Excel file into the local research database.",
+    )
+    import_parser.add_argument(
+        "--template",
+        required=True,
+        help="Template name, for example area_monthly_metrics.",
+    )
+    import_parser.add_argument(
+        "--file",
+        type=Path,
+        required=True,
+        help="CSV, XLSX, or XLSM file to import.",
+    )
+
     return parser
 
 
@@ -66,5 +99,38 @@ def main(argv: Optional[list[str]] = None) -> int:
         print(f"Generated report: {path}")
         return 0
 
+    if args.command == "validate-import":
+        try:
+            result = validate_tabular_file(args.template, args.file)
+        except ImportTemplateError as exc:
+            print(f"Import validation setup error: {exc}")
+            return 2
+
+        if result.valid:
+            print(
+                f"Import file is valid: template={result.template_name}, "
+                f"rows={result.row_count}"
+            )
+            return 0
+
+        print(
+            f"Import file is invalid: template={result.template_name}, "
+            f"rows={result.row_count}, errors={len(result.errors)}"
+        )
+        for error in result.errors:
+            print(f"row {error.row}, column {error.column}: {error.message}")
+        return 1
+
+    if args.command == "import-file":
+        try:
+            count = import_template_file(args.db, args.template, args.file)
+        except ImportTemplateError as exc:
+            print(f"Import failed: {exc}")
+            return 1
+        print(
+            f"Imported file: template={args.template}, rows={count}, database={args.db}"
+        )
+        return 0
+
     parser.error(f"Unknown command: {args.command}")
     return 2
diff --git a/src/shanghai_housing/db.py b/src/shanghai_housing/db.py
index 8e7fa3c..166928b 100644
--- a/src/shanghai_housing/db.py
+++ b/src/shanghai_housing/db.py
@@ -5,12 +5,20 @@ import sqlite3
 from pathlib import Path
 from typing import Iterable, Mapping, Union
 
+from .import_templates import ImportTemplateError, load_tabular_rows, validate_import_rows
+
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
 DEFAULT_DB_PATH = PROJECT_ROOT / "data" / "shanghai_housing.sqlite"
 SCHEMA_PATH = Path(__file__).with_name("schema.sql")
 SAMPLE_DIR = PROJECT_ROOT / "data" / "sample"
 
+SQLITE_IMPORT_TABLES = {
+    "area_monthly_metrics": "area_monthly_metrics",
+    "neighborhood_monthly_metrics": "neighborhood_monthly_metrics",
+    "policy_events": "policy_events",
+}
+
 
 def connect(db_path: Union[Path, str] = DEFAULT_DB_PATH) -> sqlite3.Connection:
     conn = sqlite3.connect(str(db_path))
@@ -73,6 +81,44 @@ def load_csv(conn: sqlite3.Connection, table: str, path: Path) -> int:
     return len(rows)
 
 
+def import_template_file(
+    db_path: Union[Path, str],
+    template_name: str,
+    file_path: Path,
+) -> int:
+    table = SQLITE_IMPORT_TABLES.get(template_name)
+    if table is None:
+        supported = ", ".join(sorted(SQLITE_IMPORT_TABLES))
+        raise ImportTemplateError(
+            f"template '{template_name}' cannot be imported into SQLite; "
+            f"supported templates: {supported}"
+        )
+
+    rows = load_tabular_rows(file_path)
+    result = validate_import_rows(template_name, rows)
+    if not result.valid:
+        first_error = result.errors[0]
+        raise ImportTemplateError(
+            f"validation failed at row {first_error.row}, "
+            f"column {first_error.column}: {first_error.message}"
+        )
+    if not rows:
+        return 0
+
+    init_db(db_path)
+    with connect(db_path) as conn:
+        columns = list(rows[0].keys())
+        placeholders = ", ".join(["?"] * len(columns))
+        column_sql = ", ".join(columns)
+        update_sql = ", ".join([f"{column}=excluded.{column}" for column in columns])
+        sql = (
+            f"INSERT INTO {table} ({column_sql}) VALUES ({placeholders}) "
+            f"ON CONFLICT DO UPDATE SET {update_sql}"
+        )
+        conn.executemany(sql, [tuple(row[column] for column in columns) for row in rows])
+    return len(rows)
+
+
 def fetch_area_metrics(conn: sqlite3.Connection, month: str) -> list[sqlite3.Row]:
     return conn.execute(
         """
diff --git a/src/shanghai_housing/import_templates.py b/src/shanghai_housing/import_templates.py
new file mode 100644
index 0000000..4af4ac7
--- /dev/null
+++ b/src/shanghai_housing/import_templates.py
@@ -0,0 +1,306 @@
+from __future__ import annotations
+
+import csv
+import json
+import re
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+from typing import Any, Mapping
+from urllib.parse import urlparse
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+IMPORT_TEMPLATE_SPEC_PATH = PROJECT_ROOT / "config" / "import_templates.json"
+
+
+@dataclass(frozen=True)
+class ImportValidationError:
+    row: int
+    column: str
+    message: str
+
+
+@dataclass(frozen=True)
+class ImportValidationResult:
+    template_name: str
+    row_count: int
+    valid: bool
+    errors: list[ImportValidationError]
+
+
+class ImportTemplateError(ValueError):
+    pass
+
+
+def load_import_template_specs(
+    path: Path = IMPORT_TEMPLATE_SPEC_PATH,
+) -> dict[str, dict[str, Any]]:
+    with path.open("r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def list_template_names(path: Path = IMPORT_TEMPLATE_SPEC_PATH) -> list[str]:
+    return sorted(load_import_template_specs(path).keys())
+
+
+def get_template_spec(
+    template_name: str,
+    path: Path = IMPORT_TEMPLATE_SPEC_PATH,
+) -> dict[str, Any]:
+    specs = load_import_template_specs(path)
+    try:
+        return specs[template_name]
+    except KeyError as exc:
+        valid_names = ", ".join(sorted(specs.keys()))
+        raise ImportTemplateError(
+            f"unknown template '{template_name}', expected one of: {valid_names}"
+        ) from exc
+
+
+def validate_csv_file(
+    template_name: str,
+    file_path: Path,
+    spec_path: Path = IMPORT_TEMPLATE_SPEC_PATH,
+) -> ImportValidationResult:
+    rows = read_csv_rows(file_path)
+    return validate_import_rows(template_name, rows, spec_path=spec_path)
+
+
+def validate_tabular_file(
+    template_name: str,
+    file_path: Path,
+    spec_path: Path = IMPORT_TEMPLATE_SPEC_PATH,
+) -> ImportValidationResult:
+    suffix = file_path.suffix.lower()
+    if suffix == ".csv":
+        rows = read_csv_rows(file_path)
+    elif suffix in {".xlsx", ".xlsm"}:
+        rows = read_excel_rows(file_path)
+    else:
+        raise ImportTemplateError("file must be a .csv, .xlsx, or .xlsm document")
+    return validate_import_rows(template_name, rows, spec_path=spec_path)
+
+
+def load_tabular_rows(file_path: Path) -> list[dict[str, str]]:
+    suffix = file_path.suffix.lower()
+    if suffix == ".csv":
+        return read_csv_rows(file_path)
+    if suffix in {".xlsx", ".xlsm"}:
+        return read_excel_rows(file_path)
+    raise ImportTemplateError("file must be a .csv, .xlsx, or .xlsm document")
+
+
+def read_csv_rows(file_path: Path) -> list[dict[str, str]]:
+    with file_path.open("r", encoding="utf-8-sig", newline="") as handle:
+        reader = csv.DictReader(handle)
+        return [dict(row) for row in reader]
+
+
+def read_excel_rows(file_path: Path) -> list[dict[str, str]]:
+    try:
+        from openpyxl import load_workbook
+    except ImportError as exc:
+        raise ImportTemplateError(
+            "Excel validation requires openpyxl; install project dependencies first"
+        ) from exc
+
+    workbook = load_workbook(file_path, read_only=True, data_only=True)
+    worksheet = workbook.active
+    rows = worksheet.iter_rows(values_only=True)
+    try:
+        headers = [str(value).strip() if value is not None else "" for value in next(rows)]
+    except StopIteration:
+        return []
+
+    records: list[dict[str, str]] = []
+    for values in rows:
+        if values is None or all(value is None or str(value).strip() == "" for value in values):
+            continue
+        record = {
+            header: "" if value is None else str(value).strip()
+            for header, value in zip(headers, values)
+            if header
+        }
+        records.append(record)
+    return records
+
+
+def validate_import_rows(
+    template_name: str,
+    rows: list[Mapping[str, Any]],
+    spec_path: Path = IMPORT_TEMPLATE_SPEC_PATH,
+) -> ImportValidationResult:
+    spec = get_template_spec(template_name, spec_path)
+    columns = spec.get("columns", [])
+    expected_columns = [column["name"] for column in columns]
+    required_columns = [
+        column["name"] for column in columns if bool(column.get("required", False))
+    ]
+    errors: list[ImportValidationError] = []
+
+    seen_columns: set[str] = set()
+    if rows:
+        seen_columns = set(rows[0].keys())
+    missing_headers = [column for column in expected_columns if column not in seen_columns]
+    unexpected_headers = sorted(seen_columns - set(expected_columns))
+
+    for column in missing_headers:
+        errors.append(ImportValidationError(0, column, "missing required header"))
+    for column in unexpected_headers:
+        errors.append(ImportValidationError(0, column, "unexpected header"))
+
+    if not rows:
+        errors.append(ImportValidationError(0, "*", "file has no data rows"))
+
+    for index, row in enumerate(rows, start=2):
+        for column in columns:
+            name = column["name"]
+            value = normalize_value(row.get(name))
+            if name in required_columns and value == "":
+                errors.append(ImportValidationError(index, name, "value is required"))
+                continue
+            if value == "":
+                continue
+            errors.extend(validate_value(index, name, value, column))
+
+    return ImportValidationResult(
+        template_name=template_name,
+        row_count=len(rows),
+        valid=not errors,
+        errors=errors,
+    )
+
+
+def normalize_value(value: Any) -> str:
+    if value is None:
+        return ""
+    return str(value).strip()
+
+
+def validate_value(
+    row_index: int,
+    column_name: str,
+    value: str,
+    column: Mapping[str, Any],
+) -> list[ImportValidationError]:
+    errors: list[ImportValidationError] = []
+    if has_sensitive_text(value):
+        return [
+            ImportValidationError(
+                row_index,
+                column_name,
+                "value appears to contain credentials or a sensitive local path",
+            )
+        ]
+
+    column_type = column["type"]
+    if column_type == "text":
+        return errors
+    if column_type == "month":
+        if not is_valid_month(value):
+            errors.append(ImportValidationError(row_index, column_name, "must be YYYY-MM"))
+        return errors
+    if column_type == "date":
+        if not is_valid_date(value):
+            errors.append(ImportValidationError(row_index, column_name, "must be YYYY-MM-DD"))
+        return errors
+    if column_type == "url":
+        if not is_valid_url(value):
+            errors.append(ImportValidationError(row_index, column_name, "must be http(s) URL"))
+        return errors
+    if column_type == "enum":
+        allowed_values = column.get("allowed_values", [])
+        if value not in allowed_values:
+            allowed = ", ".join(allowed_values)
+            errors.append(
+                ImportValidationError(row_index, column_name, f"must be one of: {allowed}")
+            )
+        return errors
+    if column_type == "integer":
+        parsed = parse_integer(value)
+        if parsed is None:
+            errors.append(ImportValidationError(row_index, column_name, "must be an integer"))
+            return errors
+        errors.extend(validate_numeric_range(row_index, column_name, parsed, column))
+        return errors
+    if column_type == "decimal":
+        parsed = parse_decimal(value)
+        if parsed is None:
+            errors.append(ImportValidationError(row_index, column_name, "must be a number"))
+            return errors
+        errors.extend(validate_numeric_range(row_index, column_name, parsed, column))
+        return errors
+
+    errors.append(
+        ImportValidationError(row_index, column_name, f"unknown type '{column_type}'")
+    )
+    return errors
+
+
+def validate_numeric_range(
+    row_index: int,
+    column_name: str,
+    value: float,
+    column: Mapping[str, Any],
+) -> list[ImportValidationError]:
+    errors: list[ImportValidationError] = []
+    minimum = column.get("min")
+    maximum = column.get("max")
+    if minimum is not None and value < float(minimum):
+        errors.append(
+            ImportValidationError(row_index, column_name, f"must be >= {minimum}")
+        )
+    if maximum is not None and value > float(maximum):
+        errors.append(
+            ImportValidationError(row_index, column_name, f"must be <= {maximum}")
+        )
+    return errors
+
+
+def parse_integer(value: str) -> int | None:
+    if not re.fullmatch(r"-?\d+", value):
+        return None
+    return int(value)
+
+
+def parse_decimal(value: str) -> float | None:
+    try:
+        return float(value)
+    except ValueError:
+        return None
+
+
+def is_valid_month(value: str) -> bool:
+    if not re.fullmatch(r"\d{4}-\d{2}", value):
+        return False
+    month = int(value[5:])
+    return 1 <= month <= 12
+
+
+def is_valid_date(value: str) -> bool:
+    if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
+        return False
+    try:
+        date.fromisoformat(value)
+    except ValueError:
+        return False
+    return True
+
+
+def is_valid_url(value: str) -> bool:
+    parsed = urlparse(value)
+    return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
+
+
+def has_sensitive_text(value: str) -> bool:
+    lower = value.lower()
+    return (
+        "password" in lower
+        or "secret" in lower
+        or "postgres://" in lower
+        or "database_url" in lower
+        or "root_" in lower
+        or value.startswith("/Users/")
+        or value.startswith("/private/")
+    )
diff --git a/src/shanghai_housing/schema.sql b/src/shanghai_housing/schema.sql
index 0200411..49983cc 100644
--- a/src/shanghai_housing/schema.sql
+++ b/src/shanghai_housing/schema.sql
@@ -60,7 +60,8 @@ CREATE TABLE IF NOT EXISTS policy_events (
     title TEXT NOT NULL,
     impact_direction INTEGER NOT NULL CHECK (impact_direction BETWEEN -2 AND 2),
     notes TEXT DEFAULT '',
-    source_url TEXT
+    source_url TEXT,
+    source TEXT NOT NULL DEFAULT 'manual'
 );
 
 CREATE INDEX IF NOT EXISTS idx_area_metrics_month ON area_monthly_metrics(month);
diff --git a/tests/test_import_templates.py b/tests/test_import_templates.py
new file mode 100644
index 0000000..734905e
--- /dev/null
+++ b/tests/test_import_templates.py
@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import csv
+import tempfile
+import unittest
+from pathlib import Path
+
+from shanghai_housing.db import connect, import_template_file
+from shanghai_housing.import_templates import validate_csv_file, validate_import_rows
+
+
+class ImportTemplateTests(unittest.TestCase):
+    def test_validates_area_monthly_metrics_template_file(self) -> None:
+        result = validate_csv_file(
+            "area_monthly_metrics",
+            Path("docs/import_templates/area_monthly_metrics.csv"),
+        )
+
+        self.assertTrue(result.valid)
+        self.assertEqual(result.row_count, 1)
+        self.assertEqual(result.errors, [])
+
+    def test_reports_missing_required_value_and_invalid_range(self) -> None:
+        rows = [
+            {
+                "area_id": "",
+                "month": "2026-13",
+                "transaction_count": "-1",
+                "transaction_price_psm": "abc",
+                "listing_count": "10",
+                "listing_price_psm": "100",
+                "median_days_on_market": "30",
+                "rent_price_psm": "3",
+                "new_supply_units": "0",
+                "land_residential_gfa_sqm": "0",
+                "mortgage_rate_pct": "3.35",
+                "policy_signal": "3",
+                "source": "manual",
+            }
+        ]
+
+        result = validate_import_rows("area_monthly_metrics", rows)
+
+        self.assertFalse(result.valid)
+        messages = {(error.column, error.message) for error in result.errors}
+        self.assertIn(("area_id", "value is required"), messages)
+        self.assertIn(("month", "must be YYYY-MM"), messages)
+        self.assertIn(("transaction_count", "must be >= 0"), messages)
+        self.assertIn(("transaction_price_psm", "must be a number"), messages)
+        self.assertIn(("policy_signal", "must be <= 2"), messages)
+
+    def test_reports_unexpected_headers_from_csv(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            path = Path(tmpdir) / "bad.csv"
+            with path.open("w", encoding="utf-8", newline="") as handle:
+                writer = csv.DictWriter(handle, fieldnames=["area_id", "unknown"])
+                writer.writeheader()
+                writer.writerow({"area_id": "qiantan", "unknown": "value"})
+
+            result = validate_csv_file("area_monthly_metrics", path)
+
+        self.assertFalse(result.valid)
+        messages = {(error.row, error.column, error.message) for error in result.errors}
+        self.assertIn((0, "unknown", "unexpected header"), messages)
+        self.assertIn((0, "month", "missing required header"), messages)
+
+    def test_imports_valid_template_file_to_sqlite(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            db_path = Path(tmpdir) / "research.sqlite"
+            count = import_template_file(
+                db_path,
+                "policy_events",
+                Path("docs/import_templates/policy_events.csv"),
+            )
+            with connect(db_path) as conn:
+                rows = conn.execute("SELECT * FROM policy_events").fetchall()
+
+        self.assertEqual(count, 1)
+        self.assertEqual(len(rows), 1)
+        self.assertEqual(rows[0]["level"], "shanghai")
+
+
+if __name__ == "__main__":
+    unittest.main()