Skip to content

Multi-Domain Analytics Platform: Data Model Plan

The Universal Abstraction

Every domain follows the same structural pattern, with domain-specific semantics:

Entity (student, prospect, bus)
  ← has attributes from DataSources (Handshake, SIS, routing software)
  ← has Resources assigned to it (advisor, bus route, case worker)
  → has Outcomes we want to predict (placement, graduation, conversion)

The goal is a schema that makes this universal pattern strongly typed at the core, while allowing domains to define their own semantics without DB migrations.

Example domains: - Career services: Entity=Student, DataSource=Handshake (meetings, employers, applications), Resource=Advisor, Outcome=Placement/Graduation - K-12 transportation: Entity=Student, DataSource=Routing software (routes, stops), Resource=Bus, Outcome=On-time rate / maintenance risk - Nonprofit: Entity=Prospect/Donor, DataSource=CRM, Resource=Case Worker, Outcome=Donor conversion / retention

Layer 1: Core Schema (strongly typed, universal)

Organization (existing tenant)
└── Domain
    ├── EntityType        (defines "Student", "Prospect", "Bus")
    │   └── Entity        (individual records)
    ├── DataSource        (Handshake, SIS, routing software, CRM)
    │   └── DataSourceRecord   (raw imported events/records)
    ├── ResourceType      (Advisor, Bus Route, Case Worker)
    │   └── Resource      (individual advisor, specific route)
    │       └── ResourceAssignment   (Entity ↔ Resource, time-bound)
    ├── OutcomeDefinition (Placement Rate, Graduation Rate, Conversion)
    │   └── OutcomeRecord (observed outcomes per entity)
    └── PredictiveModel   (trained model metadata)
        └── Prediction    (per-entity predictions)

Core Models

class Domain(OrganizationOwnedModel):
    name = models.CharField(max_length=100)           # "career_services"
    display_name = models.CharField(max_length=200)   # "Career Services"
    schema = models.JSONField(default=dict)           # see Layer 2

class EntityType(OrganizationOwnedModel):
    domain = models.ForeignKey(Domain, on_delete=models.CASCADE)
    code = models.CharField(max_length=100)           # "student", "prospect"
    schema = models.JSONField(default=dict)           # JSON Schema for attributes

class Entity(OrganizationOwnedModel):
    entity_type = models.ForeignKey(EntityType, on_delete=models.CASCADE)
    external_id = models.CharField(max_length=255)    # ID in source system
    attributes = models.JSONField(default=dict)       # domain-specific fields
    is_active = models.BooleanField(default=True)
    class Meta:
        unique_together = [("entity_type", "external_id")]

class DataSource(OrganizationOwnedModel):
    domain = models.ForeignKey(Domain, on_delete=models.CASCADE)
    code = models.CharField(max_length=100)           # "handshake", "powerschool"
    connector_class = models.CharField(max_length=255) # dotted import path
    config = models.JSONField(default=dict)           # API keys (encrypted), endpoints

class DataSourceRecord(OrganizationOwnedModel):
    source = models.ForeignKey(DataSource, on_delete=models.CASCADE)
    entity = models.ForeignKey(Entity, on_delete=models.SET_NULL, null=True)
    record_type = models.CharField(max_length=100)    # "meeting", "maintenance_event"
    external_id = models.CharField(max_length=255)
    payload = models.JSONField()                      # raw source data (immutable)
    ingested_at = models.DateTimeField(auto_now_add=True)
    class Meta:
        unique_together = [("source", "record_type", "external_id")]

class ResourceType(OrganizationOwnedModel):
    domain = models.ForeignKey(Domain, on_delete=models.CASCADE)
    code = models.CharField(max_length=100)           # "advisor", "bus_route"
    schema = models.JSONField(default=dict)

class Resource(OrganizationOwnedModel):
    resource_type = models.ForeignKey(ResourceType, on_delete=models.CASCADE)
    external_id = models.CharField(max_length=255)
    attributes = models.JSONField(default=dict)

class ResourceAssignment(OrganizationOwnedModel):
    entity = models.ForeignKey(Entity, on_delete=models.CASCADE)
    resource = models.ForeignKey(Resource, on_delete=models.CASCADE)
    assigned_at = models.DateTimeField()
    unassigned_at = models.DateTimeField(null=True)   # null = currently active
    metadata = models.JSONField(default=dict)         # e.g. assignment reason

Layer 2: Domain Schema Registry

Each Domain carries a self-describing schema in its schema JSONField. This drives validation, UI form generation, and feature extraction - without DB migrations when domains evolve.

{
  "entity_types": {
    "student": {
      "fields": {
        "gpa": {"type": "number"},
        "major": {"type": "string"},
        "graduation_semester": {"type": "string"},
        "first_generation": {"type": "boolean"}
      }
    }
  },
  "data_sources": {
    "handshake": {
      "record_types": {
        "meeting": {
          "fields": {
            "advisor_id": {"type": "string"},
            "duration_minutes": {"type": "integer"},
            "meeting_type": {"type": "string", "enum": ["career", "resume", "mock_interview"]}
          }
        },
        "application": {
          "fields": {
            "employer_id": {"type": "string"},
            "status": {"type": "string"},
            "applied_at": {"type": "string", "format": "date-time"}
          }
        }
      }
    }
  },
  "resource_types": {
    "advisor": {
      "fields": {
        "specialty": {"type": "string"},
        "capacity": {"type": "integer"}
      }
    }
  },
  "outcomes": {
    "placement": {
      "label": "Job Placement",
      "target_field": "placed",
      "target_type": "binary",
      "observation_window_days": 180
    },
    "graduation": {
      "label": "Graduation Rate",
      "target_field": "graduated_on_time",
      "target_type": "binary",
      "observation_window_days": 365
    }
  }
}

The K-12 transportation domain defines a completely different schema (bus, route, maintenance_event records) without touching the database.

A DomainSchemaValidator service validates Entity.attributes and DataSourceRecord.payload against this schema on write.

Layer 3: Predictive Modeling Layer

class OutcomeDefinition(OrganizationOwnedModel):
    domain = models.ForeignKey(Domain, on_delete=models.CASCADE)
    code = models.CharField(max_length=100)           # "placement", "graduation"
    label = models.CharField(max_length=200)
    target_type = models.CharField(max_length=50)     # "binary", "regression", "multiclass"
    config = models.JSONField(default=dict)           # from domain schema

class OutcomeRecord(OrganizationOwnedModel):
    entity = models.ForeignKey(Entity, on_delete=models.CASCADE)
    outcome = models.ForeignKey(OutcomeDefinition, on_delete=models.CASCADE)
    value = models.FloatField()                       # 0/1 for binary, continuous for regression
    observed_at = models.DateTimeField()
    metadata = models.JSONField(default=dict)

class FeatureSet(OrganizationOwnedModel):
    """Defines which fields/sources feed into a model's training."""
    domain = models.ForeignKey(Domain, on_delete=models.CASCADE)
    outcome = models.ForeignKey(OutcomeDefinition, on_delete=models.CASCADE)
    feature_definitions = models.JSONField()          # see Feature Definitions below
    version = models.PositiveIntegerField(default=1)

class PredictiveModel(OrganizationOwnedModel):
    feature_set = models.ForeignKey(FeatureSet, on_delete=models.CASCADE)
    outcome = models.ForeignKey(OutcomeDefinition, on_delete=models.CASCADE)
    model_type = models.CharField(max_length=100)     # "gradient_boost", "logistic"
    artifact_path = models.CharField(max_length=500)  # S3 / storage path
    trained_at = models.DateTimeField()
    metrics = models.JSONField(default=dict)          # AUC, accuracy, F1
    is_active = models.BooleanField(default=False)    # one active model per outcome

class Prediction(OrganizationOwnedModel):
    entity = models.ForeignKey(Entity, on_delete=models.CASCADE)
    model = models.ForeignKey(PredictiveModel, on_delete=models.CASCADE)
    predicted_value = models.FloatField()
    confidence = models.FloatField(null=True)
    explanation = models.JSONField(default=dict)      # SHAP values, top features
    predicted_at = models.DateTimeField(auto_now_add=True)
    class Meta:
        unique_together = [("entity", "model")]

Feature Definitions

FeatureSet.feature_definitions specifies how to build a feature vector from raw data. This avoids hard-coding ETL logic per domain:

{
  "entity_attributes": ["gpa", "major", "first_generation"],
  "aggregations": [
    {
      "source": "handshake",
      "record_type": "meeting",
      "aggregation": "count",
      "alias": "meeting_count",
      "filters": {"meeting_type": "career"}
    },
    {
      "source": "handshake",
      "record_type": "application",
      "aggregation": "count",
      "alias": "application_count"
    },
    {
      "source": "handshake",
      "record_type": "meeting",
      "aggregation": "sum",
      "field": "duration_minutes",
      "alias": "total_meeting_minutes"
    }
  ],
  "resource_features": [
    {
      "resource_type": "advisor",
      "feature": "is_assigned",
      "alias": "has_advisor"
    }
  ]
}

A FeatureExtractor service reads this config and executes queries against DataSourceRecord + ResourceAssignment to build a feature matrix - domain-agnostically.

Layer 4: Data Pipeline

DataSource
  └── [Connector] pulls from external API
        └── DataSourceRecord (immutable append-only raw records)
              └── [EntityResolver] links records to Entity via external_id
                    └── [AttributeDeriver] updates Entity.attributes from records
                          └── [FeatureExtractor] builds feature vectors
                                └── [ModelRunner] scores entities → Prediction

Each stage is domain-agnostic. The domain schema and feature definitions drive behavior.

Connector Interface

class BaseConnector:
    def __init__(self, data_source: DataSource): ...
    def fetch_records(self, since: datetime) -> Iterator[dict]: ...
    def get_record_type(self, raw: dict) -> str: ...
    def get_external_id(self, raw: dict) -> str: ...
    def get_entity_external_id(self, raw: dict) -> str | None: ...

One concrete subclass per integration (HandshakeConnector, PowerSchoolConnector, etc.), registered via connector_class on DataSource.

Indexing Strategy

JSONB queries need intentional indexing or they degrade at scale:

# GIN index for arbitrary JSONB key lookups on entity attributes
Index(fields=["attributes"], name="entity_attributes_gin", opclasses=["jsonb_path_ops"])

# For high-frequency filter fields, add generated columns:
# ALTER TABLE entity ADD COLUMN gpa_indexed float GENERATED ALWAYS AS
#   ((attributes->>'gpa')::float) STORED;
# Then index that column normally.

# Partial index for active entities per type
Index(fields=["entity_type", "is_active"], condition=Q(is_active=True))

# DataSourceRecord: index by source + record_type for aggregation queries
Index(fields=["source", "record_type", "entity"])

For feature extraction aggregation queries, a materialized view per domain (refreshed on ingestion) makes model training practical at scale.

Adding a New Domain

The entire process is configuration, not code:

  1. Create a Domain record with the schema JSON describing its entity types, data sources, resource types, and outcomes
  2. Register connector classes for each DataSource
  3. Define OutcomeDefinition records
  4. Build FeatureSet definitions specifying which fields/aggregations to use
  5. Run the pipeline: ingest → extract features → train → predict

No new models, no new migrations. The schema registry and generic pipeline handle everything.

Key Tradeoffs

Decision Tradeoff
JSONB for attributes Flexible schema evolution; harder to enforce constraints, slower than typed columns for high-cardinality queries
Single DataSourceRecord table Simple ingestion; very large table at scale - partition by source_id or ingested_at
Generic FeatureSet config Domain-agnostic pipelines; complex aggregations may need escape-hatch custom SQL
One Prediction table Simple; if domains have thousands of entities and many models, partition by model_id
Domain schema as JSON Zero-migration evolution; schema validation is application-layer, not DB-layer

The EAV anti-pattern is avoided because JSONB with GIN indexing + the domain schema registry gives you the flexibility of EAV with far better query performance and type awareness.