Implementation Guide: Analyze student assessment data to identify learning gaps and at-risk students

# Sample: Export student roster from PowerSchool via API to verify data structure
curl -X GET 'https://{district-powerschool-url}/ws/v1/district/student?page=1&pagesize=50' -H 'Authorization: Bearer {access_token}' -H 'Accept: application/json' | python3 -m json.tool > sample_roster.json
# Verify key fields exist
cat sample_roster.json | python3 -c "import json,sys; data=json.load(sys.stdin); students=data.get('students',{}).get('student',[]); fields=['name','student_number','grade_level','school_enrollment']; [print(f'{f}: {sum(1 for s in students if f in str(s))}/{len(students)} records') for f in fields]"
# Export sample assessment data (Renaissance Star example via SFTP)
sftp {star-sftp-user}@{star-sftp-host}:/exports/ <<< 'ls -la'

# Check vendor compliance documentation — download and archive
mkdir -p /compliance/dpas /compliance/soc2_reports /compliance/ferpa_attestations
# Panorama DPA: request from https://www.panoramaed.com/privacy
# Clever DPA: available at https://clever.com/trust/privacy/dpa
# Create compliance tracking spreadsheet
cat > /compliance/vendor_compliance_tracker.csv << 'EOF'
Vendor,DPA_Signed,DPA_Date,SOC2_Verified,SOC2_Expiry,FERPA_Attestation,COPPA_Compliant,Data_Deletion_Policy,State_Privacy_Compliant
Panorama Education,No,,No,,No,No,No,No
Clever,No,,No,,No,No,No,No
Renaissance Learning,No,,No,,No,No,No,No
Microsoft (M365),No,,No,,No,No,No,No
EOF

# --- UniFi Controller Setup (Cloud Key or self-hosted) ---
# Adopt APs via UniFi Network Application
# SSH into each AP for initial adoption if not auto-discovered:
ssh ubnt@{ap-ip-address}
set-inform http://{unifi-controller-ip}:8080/inform

# --- FortiGate 40F Initial Configuration ---
# Connect via console cable (115200 baud) or default IP 192.168.1.99
config system interface
  edit port1
    set ip 10.0.1.1/24
    set allowaccess ping https ssh
  next
end

# Configure VLAN for admin network
config system interface
  edit admin-vlan
    set vdom root
    set ip 10.0.10.1/24
    set allowaccess ping https ssh
    set interface port1
    set vlanid 10
  next
end

# Configure CIPA-compliant web filtering
config webfilter profile
  edit "CIPA-Student-Filter"
    set web-content-log enable
    config ftgd-wf
      config filters
        edit 1
          set category 2
          set action block
        next
        # Block: Adult/Mature Content, Malware, Phishing, Proxy Avoidance
        # Categories: 2,3,4,7,59,62,83,86,88,90
      end
    end
  next
end

# Apply to student VLAN policy
config firewall policy
  edit 1
    set srcintf "student-vlan"
    set dstintf port2
    set srcaddr all
    set dstaddr all
    set action accept
    set schedule always
    set service ALL
    set utm-status enable
    set webfilter-profile "CIPA-Student-Filter"
    set ssl-ssh-profile "certificate-inspection"
    set logtraffic all
  next
end

# --- Clever API verification (for MSP monitoring) ---
curl -X GET 'https://api.clever.com/v3.0/districts/{district_id}' \
  -H 'Authorization: Bearer {clever_api_token}' \
  -H 'Accept: application/json'

# Check student count matches SIS
curl -X GET 'https://api.clever.com/v3.0/students?limit=1' \
  -H 'Authorization: Bearer {clever_api_token}' | python3 -c "import json,sys; d=json.load(sys.stdin); print(f'Total students synced: check X-Total header')"

# Panorama deployment is primarily a guided SaaS setup process.
# The MSP's role is to facilitate data integration and verify accuracy.

# --- Step 5a: Initiate Panorama Onboarding ---
# Contact: partnerships@panoramaed.com or your assigned Panorama CSM
# Provide: District name, NCES ID, SIS type, student count, assessment platforms

# --- Step 5b: Configure SIS Data Feed ---
# Option 1: Clever Secure Sync (recommended)
#   - In Panorama Admin > Integrations > Add Integration > Clever
#   - Authorize Panorama in Clever Dashboard > Applications > Add Application
#   - Map fields: student_id, grade_level, school, demographics

# Option 2: Ed-Fi ODS API Connection
#   - In Panorama Admin > Integrations > Add Integration > Ed-Fi
#   - Provide Ed-Fi ODS API URL: https://{edfi-server}/api/data/v3
#   - API Key and Secret from Ed-Fi Admin App

# Option 3: SFTP CSV Upload (fallback)
#   - Generate nightly CSV exports from SIS
#   - Upload to Panorama SFTP: sftp {district}@sftp.panoramaed.com
#   - File format: student_id,first_name,last_name,grade,school_id,enrollment_date

# --- Step 5c: Configure Assessment Data Feed ---
# For Renaissance Star:
#   - In Panorama Admin > Assessments > Add Source > Renaissance
#   - Provide Star API credentials or SFTP path
#   - Map: student_id, assessment_date, subject, scale_score, percentile_rank, SGP

# For NWEA MAP:
#   - Export from MAP Admin > Data Export > Comprehensive Student File
#   - Upload via Panorama SFTP or API integration

# --- Step 5d: Configure Early Warning Indicators ---
# In Panorama Admin > Student Success > Early Warning System:
# Indicator 1: Attendance Rate
#   - Yellow flag: < 90% attendance (Chronic absence threshold)
#   - Red flag: < 80% attendance (Severe chronic absence)
# Indicator 2: Course Grades
#   - Yellow flag: Any grade D or below
#   - Red flag: 2+ grades D/F or GPA < 1.5
# Indicator 3: Assessment Performance
#   - Yellow flag: Below 25th percentile on Star/MAP
#   - Red flag: Below 10th percentile or declining SGP (< 35)
# Indicator 4: Behavior
#   - Yellow flag: 2+ disciplinary referrals in a quarter
#   - Red flag: Suspension or 4+ referrals

# Overall Risk Tier Calculation:
# Low Risk: 0 flags
# Some Risk: 1 yellow flag
# At Risk: 2+ yellow flags OR 1 red flag
# High Risk: 2+ red flags

# --- Option A: Docker-based deployment (Recommended for most MSPs) ---
# Prerequisites: Docker Engine 20.10+, Docker Compose v2, 8GB+ RAM

# Clone Ed-Fi Docker repository
git clone https://github.com/Ed-Fi-Alliance-OSS/Ed-Fi-ODS-Docker.git
cd Ed-Fi-ODS-Docker

# Configure environment
cp .env.example .env
nano .env
# Set these values:
# POSTGRES_USER=edfi_admin
# POSTGRES_PASSWORD={strong-random-password}
# ODS_POSTGRES_HOST=ed-fi-db-ods
# ADMIN_POSTGRES_HOST=ed-fi-db-admin
# API_MODE=SharedInstance
# LOGS_FOLDER=/var/log/edfi

# Launch the stack
docker compose -f docker-compose-ods-api.yml up -d

# Verify API is running
curl -s http://localhost:8001 | python3 -m json.tool
# Expected: JSON with version info and data model URLs

# --- Option B: Windows Server deployment ---
# Download installer from https://techdocs.ed-fi.org/
# EdFi.Suite3.Installer.WebApi.5.4.0.nupkg
# Run PowerShell installer:
Import-Module ./EdFi.Suite3.Installer.WebApi
$params = @{
    DbConnectionInfo = @{
        Server = '(local)'
        DatabaseName = 'EdFi_ODS'
        UseIntegratedSecurity = $true
    }
    WebSitePath = 'C:\inetpub\Ed-Fi\WebApi'
    WebSitePort = 443
    CertThumbprint = '{ssl-cert-thumbprint}'
}
Install-EdFiOdsWebApi @params

# --- Configure SIS-to-Ed-Fi sync ---
# For PowerSchool: Use Ed-Fi PowerSchool Connector
# Download from Ed-Fi Exchange: https://exchange.ed-fi.org
# Configure in connector settings:
# SIS_BASE_URL=https://{powerschool-url}/ws/v1
# EDFI_API_URL=https://{edfi-server}/api/data/v3
# EDFI_API_KEY={api-key-from-admin-app}
# EDFI_API_SECRET={api-secret-from-admin-app}

# Run initial full sync
./run-connector.sh --mode full

# Schedule nightly incremental sync via cron
echo '0 2 * * * cd /opt/edfi-connector && ./run-connector.sh --mode incremental >> /var/log/edfi-sync.log 2>&1' | crontab -

# --- Power BI Desktop: Connect to Ed-Fi ODS PostgreSQL ---
# 1. Open Power BI Desktop (free download from Microsoft)
# 2. Get Data > PostgreSQL database
#    Server: {edfi-server-ip}:5432
#    Database: EdFi_ODS
#    Data Connectivity mode: DirectQuery (for real-time) or Import (for performance)

# 3. Load key tables:
#    - edfi.Student
#    - edfi.StudentSchoolAssociation
#    - edfi.StudentSectionAssociation
#    - edfi.Grade
#    - edfi.StudentAssessment
#    - edfi.StudentAssessmentStudentObjectiveAssessment
#    - edfi.AttendanceEvent (or DailyAttendance)
#    - edfi.DisciplineAction

# --- Alternative: Connect directly to PowerSchool Oracle/MSSQL ---
# If Ed-Fi ODS is not deployed, use PowerSchool's reporting database:
# ODBC connection to PowerSchool Oracle DB (read replica)
# Tables: Students, Attendance, StoredGrades, CC (course enrollments)

# --- Power BI DAX Measures for At-Risk Scoring ---
# Create these in Power BI Desktop > Modeling > New Measure:

# Attendance Rate per Student:
AttendanceRate = 
  DIVIDE(
    COUNTROWS(FILTER(Attendance, Attendance[EventCategory] = "Present")),
    COUNTROWS(Attendance),
    0
  )

# Chronic Absence Flag:
IsChronicallyAbsent = IF([AttendanceRate] < 0.90, 1, 0)

# Assessment Risk Flag:
AssessmentRisk = 
  IF(
    CALCULATE(MAX(StudentAssessment[PercentileRank])) < 25,
    IF(CALCULATE(MAX(StudentAssessment[PercentileRank])) < 10, "High", "Moderate"),
    "Low"
  )

# --- Publish to Power BI Service ---
# 1. Publish from Desktop to your M365 tenant workspace
# 2. Configure scheduled refresh (e.g., daily at 4 AM)
# 3. Set up Row-Level Security (RLS):
#    - Role: Teacher — filter StudentSectionAssociation[StaffUSI] = USERNAME()
#    - Role: Principal — filter StudentSchoolAssociation[SchoolId] = {school_id}
#    - Role: District Admin — no filter (sees all)
# 4. Share dashboard via Power BI app or embed in Microsoft Teams

SELECT StudentUniqueId, FirstName, LastSurname, RiskLevel
         FROM vw_StudentRiskScores
         WHERE RiskLevel = 'High'
         AND RiskLevelChangedDate = CURRENT_DATE - 1

psql -h localhost -U edfi_admin -d EdFi_ODS << 'EOSQL'
CREATE OR REPLACE VIEW vw_StudentRiskScores AS
SELECT
  s.StudentUniqueId,
  s.FirstName,
  s.LastSurname,
  ssa.SchoolId,
  g.GradeLevel,
  ROUND(att.AttendanceRate * 100, 1) AS AttendancePct,
  grades.GPA,
  assess.LatestPercentile,
  CASE
    WHEN att.AttendanceRate < 0.80 OR grades.GPA < 1.5 OR assess.LatestPercentile < 10
      THEN 'High'
    WHEN att.AttendanceRate < 0.90 OR grades.FailingCount > 0 OR assess.LatestPercentile < 25
      THEN 'Moderate'
    ELSE 'Low'
  END AS RiskLevel
FROM edfi.Student s
JOIN edfi.StudentSchoolAssociation ssa ON s.StudentUSI = ssa.StudentUSI
LEFT JOIN LATERAL (
  SELECT GradeLevelDescriptorId FROM edfi.StudentSchoolAssociation
  WHERE StudentUSI = s.StudentUSI ORDER BY EntryDate DESC LIMIT 1
  ) g ON true
LEFT JOIN LATERAL (
  -- Calculate attendance rate
  SELECT
    COUNT(CASE WHEN ae.AttendanceEventCategoryDescriptorId IN (SELECT DescriptorId FROM edfi.Descriptor WHERE CodeValue = 'In Attendance') THEN 1 END)::float / NULLIF(COUNT(*), 0) AS AttendanceRate
  FROM edfi.StudentSchoolAttendanceEvent ae
  WHERE ae.StudentUSI = s.StudentUSI
    AND ae.EventDate >= CURRENT_DATE - INTERVAL '90 days'
  ) att ON true
LEFT JOIN LATERAL (
  -- Calculate GPA and failing count
  SELECT
    AVG(CASE WHEN NumericGradeEarned IS NOT NULL THEN NumericGradeEarned END) / 25.0 AS GPA,
    COUNT(CASE WHEN LetterGradeEarned IN ('D','F') THEN 1 END) AS FailingCount
  FROM edfi.Grade gr
  WHERE gr.StudentUSI = s.StudentUSI
  ) grades ON true
LEFT JOIN LATERAL (
  -- Get latest assessment percentile
  SELECT PercentileRank AS LatestPercentile
  FROM edfi.StudentAssessmentPerformanceLevel sapl
  JOIN edfi.StudentAssessment sa ON sa.StudentAssessmentIdentifier = sapl.StudentAssessmentIdentifier
  WHERE sa.StudentUSI = s.StudentUSI
  ORDER BY sa.AdministrationDate DESC LIMIT 1
  ) assess ON true;
EOSQL

# Student count reconciliation, assessment spot-check, risk score
# validation, and alert delivery test

# --- Data Validation Queries ---

# 1. Student Count Reconciliation
# Compare SIS total vs. Clever sync vs. Panorama imported
echo 'SIS Student Count:' && curl -s 'https://{powerschool-url}/ws/v1/district/student/count' -H 'Authorization: Bearer {token}'
echo 'Clever Student Count:' && curl -s 'https://api.clever.com/v3.0/students?count=true' -H 'Authorization: Bearer {clever_token}'
# Panorama: Check Admin Dashboard > Data > Import Summary

# 2. Assessment Data Spot-Check
# Pull 10 random students and verify their Star scores in Panorama match Star platform
psql -h localhost -U edfi_admin -d EdFi_ODS -c "
  SELECT s.StudentUniqueId, s.FirstName, s.LastSurname,
    sa.AdministrationDate, sasr.Result
  FROM edfi.Student s
  JOIN edfi.StudentAssessment sa ON s.StudentUSI = sa.StudentUSI
  JOIN edfi.StudentAssessmentScoreResult sasr ON sa.StudentAssessmentIdentifier = sasr.StudentAssessmentIdentifier
  ORDER BY RANDOM() LIMIT 10;
"

# 3. Risk Score Validation
# Export Panorama's at-risk list and compare with counselors' existing watch lists
# In Panorama: Reports > Early Warning > Export CSV
# Compare against the counselor's manually-maintained spreadsheet

# 4. Alert Delivery Test
# Manually trigger a test alert by temporarily lowering a test student's threshold
# Verify email delivery to the designated teacher within 15 minutes

# student_risk_model.py
# Student At-Risk Prediction Pipeline
# Requirements: pip install pandas scikit-learn sqlalchemy psycopg2-binary joblib

import pandas as pd
import numpy as np
from sqlalchemy import create_engine
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.metrics import classification_report, roc_auc_score
from sklearn.pipeline import Pipeline
import joblib
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# --- Configuration ---
DB_CONNECTION = 'postgresql://edfi_admin:{password}@{server}:5432/EdFi_ODS'
MODEL_OUTPUT_PATH = '/opt/student-analytics/models/'
PREDICTION_OUTPUT_PATH = '/opt/student-analytics/predictions/'

# Risk prediction features
FEATURES = [
    'attendance_rate_90d',      # Attendance rate over last 90 days
    'attendance_rate_30d',      # Attendance rate over last 30 days
    'attendance_trend',         # Slope of weekly attendance over 90 days
    'gpa_current',              # Current cumulative GPA
    'gpa_previous_term',        # Previous term GPA
    'gpa_trend',                # GPA change from previous term
    'failing_course_count',     # Number of courses with D or F
    'assessment_percentile',    # Latest standardized assessment percentile
    'assessment_growth',        # Student Growth Percentile (SGP)
    'discipline_count_90d',     # Discipline incidents in last 90 days
    'grade_level_numeric',      # Numeric grade level (K=0, 1-12)
    'days_enrolled',            # Days enrolled in current school
    'course_load',              # Number of active course sections
    'is_sped',                  # Special education flag (0/1)
    'is_ell',                   # English Language Learner flag (0/1)
    'is_frl',                   # Free/Reduced Lunch flag (0/1)
]

TARGET = 'became_at_risk_next_period'  # Binary: 1 = student failed a course or became chronically absent

def extract_features(engine, as_of_date=None):
    """Extract student features from Ed-Fi ODS."""
    if as_of_date is None:
        as_of_date = datetime.now().strftime('%Y-%m-%d')
    
    query = f"""
    WITH student_base AS (
        SELECT
            s.StudentUSI,
            s.StudentUniqueId,
            s.FirstName,
            s.LastSurname,
            ssa.SchoolId,
            ssa.EntryDate,
            ('{as_of_date}'::date - ssa.EntryDate) AS days_enrolled,
            COALESCE(
                (SELECT COUNT(*) FROM edfi.StudentSectionAssociation sect
                 WHERE sect.StudentUSI = s.StudentUSI
                 AND sect.EndDate IS NULL), 0
            ) AS course_load
        FROM edfi.Student s
        JOIN edfi.StudentSchoolAssociation ssa ON s.StudentUSI = ssa.StudentUSI
        WHERE ssa.ExitWithdrawDate IS NULL
    ),
    attendance_90d AS (
        SELECT
            StudentUSI,
            COUNT(CASE WHEN AttendanceEventCategoryDescriptorId IN (
                SELECT DescriptorId FROM edfi.Descriptor WHERE CodeValue = 'In Attendance'
            ) THEN 1 END)::float / NULLIF(COUNT(*), 0) AS attendance_rate_90d
        FROM edfi.StudentSchoolAttendanceEvent
        WHERE EventDate >= '{as_of_date}'::date - INTERVAL '90 days'
          AND EventDate <= '{as_of_date}'::date
        GROUP BY StudentUSI
    ),
    attendance_30d AS (
        SELECT
            StudentUSI,
            COUNT(CASE WHEN AttendanceEventCategoryDescriptorId IN (
                SELECT DescriptorId FROM edfi.Descriptor WHERE CodeValue = 'In Attendance'
            ) THEN 1 END)::float / NULLIF(COUNT(*), 0) AS attendance_rate_30d
        FROM edfi.StudentSchoolAttendanceEvent
        WHERE EventDate >= '{as_of_date}'::date - INTERVAL '30 days'
          AND EventDate <= '{as_of_date}'::date
        GROUP BY StudentUSI
    ),
    grades_current AS (
        SELECT
            StudentUSI,
            AVG(NumericGradeEarned) / 25.0 AS gpa_current,
            COUNT(CASE WHEN LetterGradeEarned IN ('D','F') THEN 1 END) AS failing_course_count
        FROM edfi.Grade
        WHERE GradingPeriodDescriptorId = (
            SELECT DescriptorId FROM edfi.Descriptor
            WHERE CodeValue = 'First Semester' -- Adjust per grading period
            LIMIT 1
        )
        GROUP BY StudentUSI
    ),
    latest_assessment AS (
        SELECT DISTINCT ON (sa.StudentUSI)
            sa.StudentUSI,
            sasr.Result::float AS assessment_percentile
        FROM edfi.StudentAssessment sa
        JOIN edfi.StudentAssessmentScoreResult sasr
            ON sa.StudentAssessmentIdentifier = sasr.StudentAssessmentIdentifier
        WHERE sasr.AssessmentReportingMethodDescriptorId = (
            SELECT DescriptorId FROM edfi.Descriptor
            WHERE CodeValue = 'Percentile' LIMIT 1
        )
        ORDER BY sa.StudentUSI, sa.AdministrationDate DESC
    ),
    discipline AS (
        SELECT
            StudentUSI,
            COUNT(*) AS discipline_count_90d
        FROM edfi.DisciplineAction
        WHERE DisciplineDate >= '{as_of_date}'::date - INTERVAL '90 days'
        GROUP BY StudentUSI
    )
    SELECT
        sb.StudentUSI,
        sb.StudentUniqueId,
        sb.FirstName,
        sb.LastSurname,
        sb.SchoolId,
        COALESCE(a90.attendance_rate_90d, 1.0) AS attendance_rate_90d,
        COALESCE(a30.attendance_rate_30d, 1.0) AS attendance_rate_30d,
        COALESCE(a30.attendance_rate_30d, 1.0) - COALESCE(a90.attendance_rate_90d, 1.0) AS attendance_trend,
        COALESCE(gc.gpa_current, 2.5) AS gpa_current,
        2.5 AS gpa_previous_term,  -- Placeholder: join to prior grading period
        COALESCE(gc.gpa_current, 2.5) - 2.5 AS gpa_trend,
        COALESCE(gc.failing_course_count, 0) AS failing_course_count,
        COALESCE(la.assessment_percentile, 50.0) AS assessment_percentile,
        50.0 AS assessment_growth,  -- Placeholder: calculate SGP from consecutive assessments
        COALESCE(d.discipline_count_90d, 0) AS discipline_count_90d,
        sb.days_enrolled,
        sb.course_load
    FROM student_base sb
    LEFT JOIN attendance_90d a90 ON sb.StudentUSI = a90.StudentUSI
    LEFT JOIN attendance_30d a30 ON sb.StudentUSI = a30.StudentUSI
    LEFT JOIN grades_current gc ON sb.StudentUSI = gc.StudentUSI
    LEFT JOIN latest_assessment la ON sb.StudentUSI = la.StudentUSI
    LEFT JOIN discipline d ON sb.StudentUSI = d.StudentUSI;
    """
    
    df = pd.read_sql(query, engine)
    logger.info(f'Extracted features for {len(df)} students as of {as_of_date}')
    return df

def train_model(engine):
    """Train the risk prediction model on historical data."""
    logger.info('Starting model training...')
    
    # Extract historical features (would need labeled historical data)
    df = extract_features(engine)
    
    # For initial deployment, use rule-based labeling as proxy:
    # A student is 'at risk' if attendance < 90% OR GPA < 2.0 OR failing_course_count > 0
    df['became_at_risk_next_period'] = (
        (df['attendance_rate_90d'] < 0.90) |
        (df['gpa_current'] < 2.0) |
        (df['failing_course_count'] > 0)
    ).astype(int)
    
    feature_cols = [
        'attendance_rate_90d', 'attendance_rate_30d', 'attendance_trend',
        'gpa_current', 'gpa_trend', 'failing_course_count',
        'assessment_percentile', 'assessment_growth',
        'discipline_count_90d', 'days_enrolled', 'course_load'
    ]
    
    X = df[feature_cols].fillna(0)
    y = df['became_at_risk_next_period']
    
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )
    
    pipeline = Pipeline([
        ('scaler', StandardScaler()),
        ('classifier', GradientBoostingClassifier(
            n_estimators=200,
            max_depth=4,
            learning_rate=0.1,
            min_samples_leaf=20,
            subsample=0.8,
            random_state=42
        ))
    ])
    
    # Cross-validation
    cv_scores = cross_val_score(pipeline, X_train, y_train, cv=5, scoring='roc_auc')
    logger.info(f'Cross-validation AUC: {cv_scores.mean():.3f} (+/- {cv_scores.std():.3f})')
    
    # Final training
    pipeline.fit(X_train, y_train)
    
    # Evaluation
    y_pred = pipeline.predict(X_test)
    y_proba = pipeline.predict_proba(X_test)[:, 1]
    
    logger.info(f'Test AUC: {roc_auc_score(y_test, y_proba):.3f}')
    logger.info(f'\n{classification_report(y_test, y_pred)}')
    
    # Feature importance
    importances = pipeline.named_steps['classifier'].feature_importances_
    for feat, imp in sorted(zip(feature_cols, importances), key=lambda x: -x[1]):
        logger.info(f'  {feat}: {imp:.3f}')
    
    # Save model
    model_path = f'{MODEL_OUTPUT_PATH}risk_model_{datetime.now().strftime("%Y%m%d")}.pkl'
    joblib.dump(pipeline, model_path)
    joblib.dump(feature_cols, f'{MODEL_OUTPUT_PATH}feature_cols.pkl')
    logger.info(f'Model saved to {model_path}')
    
    return pipeline, feature_cols

def predict_risk(engine, model=None, feature_cols=None):
    """Generate risk predictions for all current students."""
    if model is None:
        model = joblib.load(f'{MODEL_OUTPUT_PATH}risk_model_latest.pkl')
        feature_cols = joblib.load(f'{MODEL_OUTPUT_PATH}feature_cols.pkl')
    
    df = extract_features(engine)
    X = df[feature_cols].fillna(0)
    
    df['risk_probability'] = model.predict_proba(X)[:, 1]
    df['risk_tier'] = pd.cut(
        df['risk_probability'],
        bins=[0, 0.25, 0.50, 0.75, 1.0],
        labels=['Low', 'Some Risk', 'At Risk', 'High Risk']
    )
    
    # Output predictions
    output_file = f'{PREDICTION_OUTPUT_PATH}risk_scores_{datetime.now().strftime("%Y%m%d")}.csv'
    df[['StudentUniqueId', 'FirstName', 'LastSurname', 'SchoolId',
        'risk_probability', 'risk_tier',
        'attendance_rate_90d', 'gpa_current', 'failing_course_count',
        'assessment_percentile', 'discipline_count_90d'
    ]].to_csv(output_file, index=False)
    
    logger.info(f'Predictions saved to {output_file}')
    logger.info(f'Risk distribution:\n{df["risk_tier"].value_counts()}')
    
    return df

if __name__ == '__main__':
    engine = create_engine(DB_CONNECTION)
    model, feature_cols = train_model(engine)
    predictions = predict_risk(engine, model, feature_cols)
    print(f'\nTop 20 Highest Risk Students:')
    print(predictions.nlargest(20, 'risk_probability')[[
        'StudentUniqueId', 'FirstName', 'LastSurname',
        'risk_probability', 'risk_tier'
    ]].to_string(index=False))

0 3 * * 1 cd /opt/student-analytics && python3 student_risk_model.py >> /var/log/student-risk.log 2>&1

# Learning Gap Analysis Agent using OpenAI GPT-5.4

# learning_gap_agent.py
# Learning Gap Analysis Agent using OpenAI GPT-5.4
# Requirements: pip install openai pandas sqlalchemy psycopg2-binary

import os
import json
import pandas as pd
from sqlalchemy import create_engine
from openai import OpenAI
from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- Configuration ---
OPENAI_API_KEY = os.environ.get('OPENAI_API_KEY', '{your-api-key}')
DB_CONNECTION = os.environ.get('EDFI_DB_URL', 'postgresql://edfi_admin:{password}@{server}:5432/EdFi_ODS')

client = OpenAI(api_key=OPENAI_API_KEY)

SYSTEM_PROMPT = """You are an expert educational data analyst and instructional coach.
You analyze student assessment data to identify specific learning gaps and provide
actionable, teacher-friendly recommendations.

Rules:
1. Always reference specific learning standards (e.g., CCSS.MATH.4.NBT.5) when identifying gaps.
2. Prioritize gaps by impact: focus on foundational skills that block future learning.
3. Provide 2-3 specific, practical instructional strategies for each gap.
4. Use encouraging, strengths-based language (highlight what students CAN do).
5. For classroom-level analysis, identify patterns and suggest small-group configurations.
6. Never include student names in any output that could be shared beyond the teacher.
7. Format output as structured markdown for easy reading.
8. All recommendations must be grade-level appropriate and aligned to standards.
"""

def get_student_assessment_data(engine, student_id):
    """Pull detailed assessment results by learning standard for a student."""
    query = f"""
    SELECT
        a.AssessmentTitle,
        a.AssessedGradeLevelDescriptorId,
        sa.AdministrationDate,
        oa.IdentificationCode AS standard_code,
        oa.Description AS standard_description,
        saoa.Result AS score,
        saoa.ResultDatatypeTypeDescriptorId
    FROM edfi.StudentAssessment sa
    JOIN edfi.Assessment a ON sa.AssessmentIdentifier = a.AssessmentIdentifier
    LEFT JOIN edfi.StudentAssessmentStudentObjectiveAssessment saoa
        ON sa.StudentAssessmentIdentifier = saoa.StudentAssessmentIdentifier
    LEFT JOIN edfi.ObjectiveAssessment oa
        ON saoa.IdentificationCode = oa.IdentificationCode
        AND saoa.AssessmentIdentifier = oa.AssessmentIdentifier
    WHERE sa.StudentUniqueId = '{student_id}'
    ORDER BY sa.AdministrationDate DESC, oa.IdentificationCode;
    """
    return pd.read_sql(query, engine)

def get_classroom_assessment_data(engine, section_id):
    """Pull assessment results for all students in a section."""
    query = f"""
    SELECT
        s.StudentUniqueId,
        s.FirstName,
        a.AssessmentTitle,
        sa.AdministrationDate,
        oa.IdentificationCode AS standard_code,
        oa.Description AS standard_description,
        saoa.Result AS score
    FROM edfi.StudentSectionAssociation sect
    JOIN edfi.Student s ON sect.StudentUSI = s.StudentUSI
    JOIN edfi.StudentAssessment sa ON s.StudentUSI = sa.StudentUSI
    JOIN edfi.Assessment a ON sa.AssessmentIdentifier = a.AssessmentIdentifier
    LEFT JOIN edfi.StudentAssessmentStudentObjectiveAssessment saoa
        ON sa.StudentAssessmentIdentifier = saoa.StudentAssessmentIdentifier
    LEFT JOIN edfi.ObjectiveAssessment oa
        ON saoa.IdentificationCode = oa.IdentificationCode
    WHERE sect.SectionIdentifier = '{section_id}'
      AND sect.EndDate IS NULL
    ORDER BY oa.IdentificationCode, s.StudentUniqueId;
    """
    return pd.read_sql(query, engine)

def analyze_student_gaps(student_id, engine):
    """Generate individual student learning gap analysis."""
    df = get_student_assessment_data(engine, student_id)
    if df.empty:
        return 'No assessment data available for this student.'
    
    # Prepare data summary for the LLM
    data_summary = df.to_dict(orient='records')
    
    user_prompt = f"""Analyze the following assessment data for Student ID: {student_id}.
Identify specific learning gaps by standard, prioritize them, and provide
actionable instructional recommendations.

Assessment Data:
{json.dumps(data_summary, indent=2, default=str)}

Please provide:
1. **Strengths Summary**: What standards/skills has this student mastered?
2. **Priority Learning Gaps**: Ranked list of gaps with specific standards
3. **Recommended Interventions**: 2-3 specific strategies per gap
4. **Suggested Resources**: Specific activities, websites, or materials
5. **Progress Monitoring Plan**: How to track improvement on each gap
"""
    
    response = client.chat.completions.create(
        model='gpt-5.4',
        messages=[
            {'role': 'system', 'content': SYSTEM_PROMPT},
            {'role': 'user', 'content': user_prompt}
        ],
        temperature=0.3,
        max_tokens=2000
    )
    
    return response.choices[0].message.content

def analyze_classroom_gaps(section_id, engine):
    """Generate classroom-level learning gap analysis with grouping suggestions."""
    df = get_classroom_assessment_data(engine, section_id)
    if df.empty:
        return 'No assessment data available for this section.'
    
    # Aggregate by standard
    standard_summary = df.groupby(['standard_code', 'standard_description']).agg(
        avg_score=('score', lambda x: pd.to_numeric(x, errors='coerce').mean()),
        student_count=('StudentUniqueId', 'nunique'),
        below_proficient=('score', lambda x: (pd.to_numeric(x, errors='coerce') < 70).sum())
    ).reset_index()
    
    # Student-level summary (anonymized for LLM)
    student_summary = df.groupby('StudentUniqueId').apply(
        lambda g: {
            'student_code': f"Student_{g.name[-4:]}",  # Last 4 chars as anonymized ID
            'standards_below_proficient': g[pd.to_numeric(g['score'], errors='coerce') < 70]['standard_code'].tolist(),
            'standards_proficient': g[pd.to_numeric(g['score'], errors='coerce') >= 70]['standard_code'].tolist()
        }
    ).tolist()
    
    user_prompt = f"""Analyze the following classroom assessment data for Section: {section_id}.

Standard-Level Summary (aggregated across all students):
{standard_summary.to_string(index=False)}

Student-Level Gap Patterns (anonymized):
{json.dumps(student_summary, indent=2, default=str)}

Please provide:
1. **Class-Wide Strengths**: Standards where most students are proficient
2. **Class-Wide Gaps**: Standards where >30% of students are below proficient
3. **Small Group Recommendations**: Suggest 3-4 skill-based groups with student codes
4. **Whole-Class Reteaching Priorities**: Top 2 standards to reteach to everyone
5. **Differentiated Instruction Plan**: Specific activities for each group
6. **Timeline**: Suggested 2-week plan to address top gaps
"""
    
    response = client.chat.completions.create(
        model='gpt-5.4',
        messages=[
            {'role': 'system', 'content': SYSTEM_PROMPT},
            {'role': 'user', 'content': user_prompt}
        ],
        temperature=0.3,
        max_tokens=3000
    )
    
    return response.choices[0].message.content

def generate_weekly_gap_report(engine, school_id):
    """Generate a weekly school-level learning gap summary for the principal."""
    query = f"""
    SELECT
        oa.IdentificationCode AS standard_code,
        oa.Description AS standard_description,
        COUNT(DISTINCT sa.StudentUSI) AS students_assessed,
        AVG(saoa.Result::numeric) AS avg_score,
        COUNT(CASE WHEN saoa.Result::numeric < 70 THEN 1 END) AS below_proficient_count,
        ROUND(COUNT(CASE WHEN saoa.Result::numeric < 70 THEN 1 END)::numeric / 
              NULLIF(COUNT(DISTINCT sa.StudentUSI), 0) * 100, 1) AS pct_below_proficient
    FROM edfi.StudentAssessment sa
    JOIN edfi.StudentSchoolAssociation ssa ON sa.StudentUSI = ssa.StudentUSI
    JOIN edfi.StudentAssessmentStudentObjectiveAssessment saoa
        ON sa.StudentAssessmentIdentifier = saoa.StudentAssessmentIdentifier
    JOIN edfi.ObjectiveAssessment oa
        ON saoa.IdentificationCode = oa.IdentificationCode
    WHERE ssa.SchoolId = {school_id}
      AND sa.AdministrationDate >= CURRENT_DATE - INTERVAL '30 days'
    GROUP BY oa.IdentificationCode, oa.Description
    HAVING COUNT(DISTINCT sa.StudentUSI) >= 5
    ORDER BY pct_below_proficient DESC;
    """
    df = pd.read_sql(query, engine)
    
    user_prompt = f"""Create a brief weekly learning gap summary for a school principal.

School-wide Assessment Gap Data (last 30 days):
{df.to_string(index=False)}

Provide:
1. **Executive Summary**: 2-3 sentences on overall school performance
2. **Top 5 Priority Standards**: Standards with highest percentage below proficient
3. **Recommended School-Wide Actions**: 2-3 actionable next steps
4. **Bright Spots**: Any standards showing strong performance to celebrate

Keep it concise — principals have 2 minutes to read this.
"""
    
    response = client.chat.completions.create(
        model='gpt-5.4',
        messages=[
            {'role': 'system', 'content': SYSTEM_PROMPT},
            {'role': 'user', 'content': user_prompt}
        ],
        temperature=0.3,
        max_tokens=1000
    )
    
    return response.choices[0].message.content

# --- CLI Interface ---
if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser(description='Learning Gap Analyzer')
    parser.add_argument('--mode', choices=['student', 'classroom', 'school'], required=True)
    parser.add_argument('--id', required=True, help='Student ID, Section ID, or School ID')
    args = parser.parse_args()
    
    engine = create_engine(DB_CONNECTION)
    
    if args.mode == 'student':
        result = analyze_student_gaps(args.id, engine)
    elif args.mode == 'classroom':
        result = analyze_classroom_gaps(args.id, engine)
    elif args.mode == 'school':
        result = generate_weekly_gap_report(engine, int(args.id))
    
    print(result)

# Individual student gap analysis
python3 learning_gap_agent.py --mode student --id 'STU12345'

# Classroom-level analysis with grouping suggestions
python3 learning_gap_agent.py --mode classroom --id 'SEC-MATH-4A'

# Weekly school principal summary
python3 learning_gap_agent.py --mode school --id 12345

# sync_monitor.py
# SIS-to-Analytics Data Sync Health Monitor
# Requirements: pip install requests psycopg2-binary schedule

import requests
import psycopg2
import json
import os
import logging
from datetime import datetime, timedelta

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/sync-monitor.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# --- Configuration ---
CONFIG = {
    'clever': {
        'api_url': 'https://api.clever.com/v3.0',
        'token': os.environ.get('CLEVER_API_TOKEN', '{token}'),
    },
    'edfi': {
        'api_url': os.environ.get('EDFI_API_URL', 'https://{server}/api/data/v3'),
        'api_key': os.environ.get('EDFI_API_KEY', '{key}'),
        'api_secret': os.environ.get('EDFI_API_SECRET', '{secret}'),
    },
    'database': {
        'host': os.environ.get('DB_HOST', 'localhost'),
        'port': 5432,
        'dbname': 'EdFi_ODS',
        'user': os.environ.get('DB_USER', 'edfi_admin'),
        'password': os.environ.get('DB_PASSWORD', '{password}'),
    },
    'psa_webhook': {
        # ConnectWise Manage API or generic webhook
        'url': os.environ.get('PSA_WEBHOOK_URL', 'https://{connectwise-url}/v4_6_release/apis/3.0/service/tickets'),
        'api_key': os.environ.get('PSA_API_KEY', '{psa-key}'),
        'company_id': os.environ.get('PSA_COMPANY_ID', '{company}'),
    },
    'thresholds': {
        'max_count_discrepancy_pct': 5.0,   # Alert if counts differ by >5%
        'max_sync_age_hours': 26,             # Alert if last sync >26 hours ago
        'min_student_count': 10,              # Alert if fewer students than expected
    },
    'alert_email': os.environ.get('ALERT_EMAIL', 'alerts@{msp-domain}.com'),
}

def get_clever_counts():
    """Get student and teacher counts from Clever API."""
    headers = {'Authorization': f'Bearer {CONFIG["clever"]["token"]}'}
    
    students = requests.get(
        f'{CONFIG["clever"]["api_url"]}/students?limit=1',
        headers=headers
    )
    teachers = requests.get(
        f'{CONFIG["clever"]["api_url"]}/teachers?limit=1',
        headers=headers
    )
    sections = requests.get(
        f'{CONFIG["clever"]["api_url"]}/sections?limit=1',
        headers=headers
    )
    
    return {
        'students': int(students.headers.get('X-Total-Count', 0)),
        'teachers': int(teachers.headers.get('X-Total-Count', 0)),
        'sections': int(sections.headers.get('X-Total-Count', 0)),
        'status': 'ok' if students.status_code == 200 else 'error',
    }

def get_edfi_counts():
    """Get student counts from Ed-Fi ODS database."""
    conn = psycopg2.connect(**CONFIG['database'])
    cur = conn.cursor()
    
    counts = {}
    queries = {
        'students': "SELECT COUNT(*) FROM edfi.Student s JOIN edfi.StudentSchoolAssociation ssa ON s.StudentUSI = ssa.StudentUSI WHERE ssa.ExitWithdrawDate IS NULL",
        'staff': "SELECT COUNT(*) FROM edfi.Staff sf JOIN edfi.StaffSchoolAssociation sfa ON sf.StaffUSI = sfa.StaffUSI",
        'sections': "SELECT COUNT(*) FROM edfi.Section WHERE id IS NOT NULL",
        'assessments': "SELECT COUNT(*) FROM edfi.StudentAssessment WHERE AdministrationDate >= CURRENT_DATE - INTERVAL '90 days'",
        'latest_change': "SELECT MAX(LastModifiedDate) FROM edfi.Student",
    }
    
    for key, query in queries.items():
        try:
            cur.execute(query)
            result = cur.fetchone()[0]
            counts[key] = result if result else 0
        except Exception as e:
            counts[key] = f'ERROR: {str(e)}'
            logger.error(f'Query failed for {key}: {e}')
    
    cur.close()
    conn.close()
    return counts

def check_sync_health():
    """Main health check: compare counts and detect issues."""
    issues = []
    
    logger.info('Starting sync health check...')
    
    # Get counts from both sources
    try:
        clever = get_clever_counts()
        logger.info(f'Clever counts: {clever}')
    except Exception as e:
        issues.append(f'CRITICAL: Cannot reach Clever API: {e}')
        clever = None
    
    try:
        edfi = get_edfi_counts()
        logger.info(f'Ed-Fi ODS counts: {edfi}')
    except Exception as e:
        issues.append(f'CRITICAL: Cannot connect to Ed-Fi ODS database: {e}')
        edfi = None
    
    # Compare student counts
    if clever and edfi and isinstance(edfi.get('students'), int):
        clever_students = clever.get('students', 0)
        edfi_students = edfi.get('students', 0)
        
        if clever_students > 0:
            discrepancy_pct = abs(clever_students - edfi_students) / clever_students * 100
            if discrepancy_pct > CONFIG['thresholds']['max_count_discrepancy_pct']:
                issues.append(
                    f'WARNING: Student count mismatch - Clever: {clever_students}, '
                    f'Ed-Fi ODS: {edfi_students} ({discrepancy_pct:.1f}% difference)'
                )
        
        if edfi_students < CONFIG['thresholds']['min_student_count']:
            issues.append(
                f'CRITICAL: Ed-Fi ODS has only {edfi_students} students - '
                f'possible sync failure or data loss'
            )
    
    # Check sync freshness
    if edfi and isinstance(edfi.get('latest_change'), datetime):
        age_hours = (datetime.now() - edfi['latest_change']).total_seconds() / 3600
        if age_hours > CONFIG['thresholds']['max_sync_age_hours']:
            issues.append(
                f'WARNING: Ed-Fi ODS data is {age_hours:.1f} hours old - '
                f'last update: {edfi["latest_change"]}. Expected nightly sync.'
            )
    
    # Check assessment data freshness
    if edfi and isinstance(edfi.get('assessments'), int):
        if edfi['assessments'] == 0:
            issues.append(
                'WARNING: No assessment records in the last 90 days - '
                'verify assessment data feed is active'
            )
    
    return issues

def create_psa_ticket(issues):
    """Create a ticket in the MSP's PSA (ConnectWise Manage example)."""
    severity = 'Critical' if any('CRITICAL' in i for i in issues) else 'Warning'
    
    ticket_data = {
        'summary': f'[Auto] Student Analytics Sync {severity} - {datetime.now().strftime("%Y-%m-%d")}',
        'initialDescription': '\n'.join(issues),
        'board': {'name': 'Education Managed Services'},
        'status': {'name': 'New'},
        'priority': {'id': 1 if severity == 'Critical' else 3},
        'company': {'id': CONFIG['psa_webhook']['company_id']},
    }
    
    # For generic webhook (Halo, Autotask, or Slack):
    webhook_payload = {
        'text': f'🚨 *Student Analytics Sync Alert* ({severity})\n' + '\n'.join(f'• {i}' for i in issues),
        'timestamp': datetime.now().isoformat(),
        'district': CONFIG['psa_webhook']['company_id'],
    }
    
    try:
        response = requests.post(
            CONFIG['psa_webhook']['url'],
            json=webhook_payload,
            headers={
                'Content-Type': 'application/json',
                'Authorization': f'Bearer {CONFIG["psa_webhook"]["api_key"]}'
            },
            timeout=30
        )
        logger.info(f'PSA ticket created: {response.status_code}')
    except Exception as e:
        logger.error(f'Failed to create PSA ticket: {e}')

def run_monitor():
    """Execute the full monitoring cycle."""
    issues = check_sync_health()
    
    if issues:
        logger.warning(f'Found {len(issues)} sync issues:')
        for issue in issues:
            logger.warning(f'  - {issue}')
        create_psa_ticket(issues)
    else:
        logger.info('All sync health checks passed ✓')
    
    # Log summary to file for trend analysis
    summary = {
        'timestamp': datetime.now().isoformat(),
        'issues_count': len(issues),
        'issues': issues,
        'status': 'ALERT' if issues else 'OK'
    }
    with open('/var/log/sync-monitor-history.jsonl', 'a') as f:
        f.write(json.dumps(summary) + '\n')

if __name__ == '__main__':
    run_monitor()

# Install dependencies
pip install requests psycopg2-binary

# Set environment variables
export CLEVER_API_TOKEN='{token}'
export EDFI_API_URL='https://{server}/api/data/v3'
export DB_HOST='localhost'
export DB_PASSWORD='{password}'
export PSA_WEBHOOK_URL='https://{psa-url}/api/tickets'
export PSA_API_KEY='{key}'

# Schedule to run every 6 hours
crontab -e
# Add: 0 */6 * * * cd /opt/student-analytics && python3 sync_monitor.py

# Assessment Data ETL Pipeline supporting Renaissance Star (SFTP/CSV), NWEA
# MAP (CSV), and iReady (CSV)

# assessment_etl.py
# Assessment Data ETL Pipeline
# Supports: Renaissance Star (SFTP/CSV), NWEA MAP (CSV), iReady (CSV)
# Requirements: pip install pandas paramiko sqlalchemy psycopg2-binary

import pandas as pd
import paramiko
import os
import glob
import logging
from sqlalchemy import create_engine
from datetime import datetime

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# --- Configuration ---
CONFIG = {
    'star': {
        'sftp_host': os.environ.get('STAR_SFTP_HOST', 'sftp.renaissance.com'),
        'sftp_user': os.environ.get('STAR_SFTP_USER', '{user}'),
        'sftp_password': os.environ.get('STAR_SFTP_PASS', '{pass}'),
        'remote_path': '/exports/',
        'local_path': '/opt/student-analytics/data/star/',
    },
    'map': {
        'import_path': '/opt/student-analytics/data/map/',
    },
    'iready': {
        'import_path': '/opt/student-analytics/data/iready/',
    },
    'database': {
        'connection': os.environ.get('DB_URL', 'postgresql://edfi_admin:{password}@localhost:5432/EdFi_ODS'),
    }
}

# Common output schema
OUTPUT_COLUMNS = [
    'student_unique_id',
    'assessment_source',      # 'Star', 'MAP', 'iReady'
    'assessment_subject',     # 'Reading', 'Math', 'EarlyLiteracy'
    'administration_date',
    'scale_score',
    'percentile_rank',
    'growth_percentile',      # SGP or equivalent
    'grade_equivalent',
    'lexile_or_quantile',     # Lexile (reading) or Quantile (math)
    'proficiency_level',      # 'Below', 'Approaching', 'At', 'Above'
    'standard_scores',        # JSON: {"CCSS.MATH.4.NBT.5": 72, ...}
]

def extract_star_data():
    """Download latest Star assessment exports via SFTP."""
    logger.info('Extracting Star assessment data via SFTP...')
    
    transport = paramiko.Transport((CONFIG['star']['sftp_host'], 22))
    transport.connect(
        username=CONFIG['star']['sftp_user'],
        password=CONFIG['star']['sftp_password']
    )
    sftp = paramiko.SFTPClient.from_transport(transport)
    
    os.makedirs(CONFIG['star']['local_path'], exist_ok=True)
    
    remote_files = sftp.listdir(CONFIG['star']['remote_path'])
    csv_files = [f for f in remote_files if f.endswith('.csv') and 'Assessment' in f]
    
    for filename in csv_files:
        remote_path = CONFIG['star']['remote_path'] + filename
        local_path = CONFIG['star']['local_path'] + filename
        sftp.get(remote_path, local_path)
        logger.info(f'  Downloaded: {filename}')
    
    sftp.close()
    transport.close()
    return CONFIG['star']['local_path']

def transform_star(data_path):
    """Transform Star CSV exports to common schema."""
    logger.info('Transforming Star data...')
    
    all_files = glob.glob(os.path.join(data_path, '*.csv'))
    dfs = []
    
    for f in all_files:
        df = pd.read_csv(f)
        # Star typical columns: StudentID, AssessmentDate, Subject,
        # ScaledScore, PercentileRank, SGP, GradeEquivalent, ...
        
        transformed = pd.DataFrame({
            'student_unique_id': df.get('StudentID', df.get('Student ID', '')).astype(str),
            'assessment_source': 'Star',
            'assessment_subject': df.get('AssessmentType', df.get('Subject', 'Unknown')),
            'administration_date': pd.to_datetime(df.get('CompletedDate', df.get('Assessment Date'))),
            'scale_score': pd.to_numeric(df.get('ScaledScore', df.get('Scaled Score')), errors='coerce'),
            'percentile_rank': pd.to_numeric(df.get('PercentileRank', df.get('Percentile Rank')), errors='coerce'),
            'growth_percentile': pd.to_numeric(df.get('StudentGrowthPercentile', df.get('SGP')), errors='coerce'),
            'grade_equivalent': df.get('GradeEquivalent', df.get('Grade Equivalent', '')).astype(str),
            'lexile_or_quantile': df.get('Lexile', df.get('Quantile', '')).astype(str),
            'proficiency_level': df.get('ProficiencyLevel', df.get('Benchmark Category', 'Unknown')),
            'standard_scores': '{}',  # Star provides domain scores, not standard-level
        })
        dfs.append(transformed)
    
    result = pd.concat(dfs, ignore_index=True) if dfs else pd.DataFrame(columns=OUTPUT_COLUMNS)
    logger.info(f'  Transformed {len(result)} Star assessment records')
    return result

def transform_map(data_path):
    """Transform NWEA MAP CSV exports to common schema."""
    logger.info('Transforming MAP data...')
    
    all_files = glob.glob(os.path.join(data_path, '*.csv'))
    dfs = []
    
    for f in all_files:
        df = pd.read_csv(f)
        # MAP typical columns: StudentID, TermName, TestRITScore,
        # TestPercentile, FallToFallProjectedGrowth, ...
        
        transformed = pd.DataFrame({
            'student_unique_id': df.get('StudentID', '').astype(str),
            'assessment_source': 'MAP',
            'assessment_subject': df.get('Discipline', df.get('Subject', 'Unknown')),
            'administration_date': pd.to_datetime(df.get('TestStartDate', df.get('Test Start Date'))),
            'scale_score': pd.to_numeric(df.get('TestRITScore', df.get('RIT Score')), errors='coerce'),
            'percentile_rank': pd.to_numeric(df.get('TestPercentile', df.get('Percentile')), errors='coerce'),
            'growth_percentile': pd.to_numeric(df.get('GrowthIndex', None), errors='coerce'),
            'grade_equivalent': '',
            'lexile_or_quantile': df.get('LexileScore', df.get('RITtoReading', '')).astype(str),
            'proficiency_level': df.get('AchievementQuintile', 'Unknown').astype(str),
            'standard_scores': '{}',
        })
        dfs.append(transformed)
    
    result = pd.concat(dfs, ignore_index=True) if dfs else pd.DataFrame(columns=OUTPUT_COLUMNS)
    logger.info(f'  Transformed {len(result)} MAP assessment records')
    return result

def load_to_database(df, engine):
    """Load transformed assessment data into the analytics database."""
    logger.info(f'Loading {len(df)} records to database...')
    
    # Deduplicate: keep latest record per student per subject per date
    df = df.drop_duplicates(
        subset=['student_unique_id', 'assessment_source', 'assessment_subject', 'administration_date'],
        keep='last'
    )
    
    # Load to staging table
    df.to_sql(
        'stg_assessment_import',
        engine,
        schema='analytics',
        if_exists='replace',
        index=False,
        method='multi',
        chunksize=1000
    )
    
    # Merge into main assessment table
    with engine.connect() as conn:
        conn.execute("""
            INSERT INTO analytics.student_assessments
            SELECT * FROM analytics.stg_assessment_import s
            WHERE NOT EXISTS (
                SELECT 1 FROM analytics.student_assessments a
                WHERE a.student_unique_id = s.student_unique_id
                AND a.assessment_source = s.assessment_source
                AND a.assessment_subject = s.assessment_subject
                AND a.administration_date = s.administration_date
            );
        """)
        conn.commit()
    
    logger.info('Data load complete ✓')

def run_etl():
    """Execute the full ETL pipeline."""
    engine = create_engine(CONFIG['database']['connection'])
    all_data = []
    
    # Star ETL
    try:
        star_path = extract_star_data()
        star_df = transform_star(star_path)
        all_data.append(star_df)
    except Exception as e:
        logger.error(f'Star ETL failed: {e}')
    
    # MAP ETL (expects manual CSV drop or automated download)
    try:
        if os.path.exists(CONFIG['map']['import_path']):
            map_df = transform_map(CONFIG['map']['import_path'])
            all_data.append(map_df)
    except Exception as e:
        logger.error(f'MAP ETL failed: {e}')
    
    # Combine and load
    if all_data:
        combined = pd.concat(all_data, ignore_index=True)
        load_to_database(combined, engine)
        logger.info(f'ETL pipeline complete: {len(combined)} total records processed')
    else:
        logger.warning('No assessment data processed in this run')

if __name__ == '__main__':
    run_etl()

-- run once to create the analytics schema and student_assessments table

CREATE SCHEMA IF NOT EXISTS analytics;

CREATE TABLE IF NOT EXISTS analytics.student_assessments (
    student_unique_id VARCHAR(50),
    assessment_source VARCHAR(20),
    assessment_subject VARCHAR(50),
    administration_date DATE,
    scale_score NUMERIC,
    percentile_rank NUMERIC,
    growth_percentile NUMERIC,
    grade_equivalent VARCHAR(10),
    lexile_or_quantile VARCHAR(20),
    proficiency_level VARCHAR(30),
    standard_scores JSONB DEFAULT '{}',
    created_at TIMESTAMP DEFAULT NOW(),
    PRIMARY KEY (student_unique_id, assessment_source, assessment_subject, administration_date)
);

CREATE INDEX idx_assessments_student ON analytics.student_assessments(student_unique_id);
CREATE INDEX idx_assessments_date ON analytics.student_assessments(administration_date);

0 4 * * * cd /opt/student-analytics && python3 assessment_etl.py >> /var/log/assessment-etl.log 2>&1