Field Validation

Field Validation and Data Quality

Validate field completeness, format correctness, and data quality for vehicle insurance CSV/Excel files.

When to Activate

Use this skill when the user:

• Says "validate the data" or "check data quality"
• Mentions "missing fields", "required fields", or "field validation"
• Asks "is the data complete?" or "quality score"
• Wants to "generate quality report"
• Mentions CSV or Excel file validation

Quick Start Workflow

When activated, follow this 3-step process:

Step 1: Load Data
  ↓
Step 2: Run Validation Checks (4 layers)
  ↓
Step 3: Generate Quality Report

Step 1: Load and Inspect Data

1.1 Identify File Location

# Project data files
data_file = 'data/车险清单_2025年10-11月_合并.csv'
mapping_file = '业务员机构团队归属.json'

1.2 Load with Pandas

import pandas as pd

# Load main data
df = pd.read_csv(data_file, encoding='utf-8-sig', low_memory=False)

# Quick inspection
print(f"Loaded {len(df)} records, {len(df.columns)} columns")
print(f"Date range: {df['投保确认时间'].min()} to {df['投保确认时间'].max()}")

Step 2: Run 4-Layer Validation

Layer 1: Required Fields (P0 - Blocking)

Check these 7 critical fields MUST exist and have values:

Validation Code:

def validate_required_fields(df):
    """P0 validation - blocking errors"""
    required = ['投保确认时间', '三级机构', '业务员', '客户类别3',
                '签单/批改保费', '签单数量', '是否续保']

    missing_cols = [col for col in required if col not in df.columns]
    if missing_cols:
        return {'valid': False, 'missing_columns': missing_cols}

    # Check null values
    null_counts = df[required].isnull().sum()
    problematic = null_counts[null_counts > 0].to_dict()

    return {
        'valid': len(problematic) == 0,
        'null_counts': problematic,
        'total_invalid_rows': df[required].isnull().any(axis=1).sum()
    }

Layer 2: Format Validation (P0 - Blocking)

Check data types and formats:

def validate_formats(df):
    """P0 validation - format errors"""
    errors = []

    # Date format
    try:
        df['投保确认时间'] = pd.to_datetime(df['投保确认时间'], errors='coerce')
        invalid_dates = df['投保确认时间'].isnull().sum()
        if invalid_dates > 0:
            errors.append(f'Invalid dates: {invalid_dates} rows')
    except:
        errors.append('Date column format error')

    # Numeric format
    try:
        df['签单/批改保费'] = pd.to_numeric(df['签单/批改保费'], errors='coerce')
        invalid_premium = df['签单/批改保费'].isnull().sum()
        if invalid_premium > 0:
            errors.append(f'Non-numeric premium: {invalid_premium} rows')
    except:
        errors.append('Premium column format error')

    return {'valid': len(errors) == 0, 'errors': errors}

Layer 3: Range Validation (P1 - Warning)

Check if values are within reasonable ranges:

def validate_ranges(df):
    """P1 validation - warnings only"""
    warnings = []

    # Premium range check
    out_of_range = df[
        (df['签单/批改保费'] < -1000000) |
        (df['签单/批改保费'] > 100000)
    ]
    if len(out_of_range) > 0:
        warnings.append(f'Extreme premium values: {len(out_of_range)} rows')

    # Policy count check
    invalid_count = df[(df['签单数量'] < 1) | (df['签单数量'] > 10000)]
    if len(invalid_count) > 0:
        warnings.append(f'Invalid policy count: {len(invalid_count)} rows')

    return {'warnings': warnings}

Layer 4: Consistency Validation (P1 - Warning)

Check data consistency with staff mapping:

def validate_consistency(df, staff_mapping):
    """P1 validation - check against mapping table"""
    import re

    # Build name lookup
    name_to_info = {}
    for staff_key, info in staff_mapping.items():
        match = re.search(r'[\u4e00-\u9fa5]+', staff_key)
        if match:
            name_to_info[match.group()] = info

    # Check unmapped staff
    data_staff = df['业务员'].unique()
    unmapped = [s for s in data_staff if s not in name_to_info]

    return {
        'unmapped_staff': unmapped[:10],  # First 10
        'unmapped_count': len(unmapped),
        'coverage_rate': 1.0 - (len(unmapped) / len(data_staff))
    }

Step 3: Generate Quality Report

3.1 Calculate Quality Score (0-100)

def calculate_quality_score(validation_results):
    """Compute overall quality score"""
    score = 0

    # Required fields (30 points)
    if validation_results['required_fields']['valid']:
        score += 30

    # Format validation (30 points)
    if validation_results['format']['valid']:
        score += 30

    # Range check (20 points)
    if len(validation_results['range']['warnings']) == 0:
        score += 20
    elif len(validation_results['range']['warnings']) <= 2:
        score += 10

    # Consistency (20 points)
    coverage = validation_results['consistency']['coverage_rate']
    if coverage >= 0.98:
        score += 20
    elif coverage >= 0.95:
        score += 15
    elif coverage >= 0.90:
        score += 10

    return score

3.2 Report Template

def generate_report(df, validation_results, score):
    """Generate human-readable quality report"""
    report = f"""
# Data Quality Report

**File**: {df.attrs.get('filename', 'Unknown')}
**Records**: {len(df):,}
**Quality Score**: {score}/100 {'✅' if score >= 90 else '⚠️' if score >= 75 else '❌'}

## Validation Results

### ✅ Passed Checks:
{_format_passed_checks(validation_results)}

### ⚠️ Warnings:
{_format_warnings(validation_results)}

### ❌ Errors:
{_format_errors(validation_results)}

## Recommendations

{_generate_recommendations(validation_results, score)}
"""
    return report

Common Use Cases

Case 1: "Validate my CSV file"

# Quick validation
df = pd.read_csv('data.csv', encoding='utf-8-sig')

results = {
    'required_fields': validate_required_fields(df),
    'format': validate_formats(df),
    'range': validate_ranges(df)
}

score = calculate_quality_score(results)
print(f"Quality Score: {score}/100")

Case 2: "Check for missing values"

# Focus on Layer 1
result = validate_required_fields(df)

if not result['valid']:
    print(f"❌ Found {result['total_invalid_rows']} rows with missing required fields")
    print(f"Null counts: {result['null_counts']}")
else:
    print("✅ All required fields are complete")

Case 3: "Is my data quality good?"

# Full validation pipeline
df = pd.read_csv('data.csv', encoding='utf-8-sig')
mapping = json.load(open('mapping.json'))

results = {
    'required_fields': validate_required_fields(df),
    'format': validate_formats(df),
    'range': validate_ranges(df),
    'consistency': validate_consistency(df, mapping)
}

score = calculate_quality_score(results)
report = generate_report(df, results, score)
print(report)

Quality Score Interpretation

Troubleshooting

"Many missing required fields"

Cause: Column names don't match expected names

Solution:

1. Check actual column names: df.columns.tolist()
2. Verify encoding: use encoding='utf-8-sig' for Excel exports
3. Check for extra spaces in column names: df.columns = df.columns.str.strip()

"Date format validation fails"

Cause: Date format inconsistency

Solution:

# Try multiple date formats
df['投保确认时间'] = pd.to_datetime(
    df['投保确认时间'],
    format='%Y-%m-%d',  # Adjust format
    errors='coerce'
)

"Low coverage rate for staff mapping"

Cause: Mapping table is outdated

Action: Alert user to update 业务员机构团队归属.json

Related Files

Main data processor: backend/data_processor.py

• Uses these validation functions
• See lines 132-156 for cleaning logic

Field definitions: docs/FIELD_MAPPING.md

• Complete field dictionary
• Data types and business rules

Related Skills:

• data-cleaning-standards - What to do after validation fails
• staff-mapping-management - How to update mapping table

Skill Version: v1.0 Created: 2025-11-09 File Size: ~330 lines Focuses On: Field validation only (not cleaning or mapping)