Differential Contrastive Integrity

The problem

It is difficult to ensure that different versions of the same data remain consistent and accurate when viewed from different perspectives.

What it does

The tool compares different views of the same data to verify that they remain consistent with one another. It outputs a simple true or false result to confirm this integrity.

Why it matters

It provides a way to verify that data remains consistent across different formats or views.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 differential_contrastive_integrity.py
Integrity Check Result: True

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 51 lines, one file, standard library only.

# Differential Contrastive Integrity Check v2
import math
from typing import Dict, Any, Tuple, Optional

def validate_schema(data: Dict[str, Any], schema: Dict[str, Any]) -> bool:
    for key, expected_type in schema.items():
        if key not in data:
            return False
        if not isinstance(data[key], expected_type):
            return False
    return True

def cosine_similarity(a: list, b: list) -> float:
    dot_product = sum(x * y for x, y in zip(a, b))
    norm_a = math.sqrt(sum(x ** 2 for x in a))
    norm_b = math.sqrt(sum(y ** 2 for y in b))
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot_product / (norm_a * norm_b)

def contrastive_loss(embedding1: list, embedding2: list, is_similar: bool) -> float:
    similarity = cosine_similarity(embedding1, embedding2)
    if is_similar:
        return 1 - similarity  # Minimize loss when similar
    return similarity  # Maximize separation when different

def differential_contrastive_integrity_check(
    data_view1: Dict[str, Any],
    data_view2: Dict[str, Any],
    schema: Dict[str, Any],
    embeddings: Dict[str, list]
) -> Tuple[bool, Optional[float]]:
    if not validate_schema(data_view1, schema) or not validate_schema(data_view2, schema):
        return (False, None)
    emb1 = embeddings.get(data_view1.get('id'), [])
    emb2 = embeddings.get(data_view2.get('id'), [])
    if not emb1 or not emb2:
        return (False, None)
    loss = contrastive_loss(emb1, emb2, is_similar=True)
    return (loss < 0.5, loss)

if __name__ == "__main__":
    schema = {'id': str, 'value': float}
    data1 = {'id': 'view1', 'value': 3.14}
    data2 = {'id': 'view2', 'value': 3.14159}
    embeddings = {
        'view1': [1.0, 0.1, 0.2],
        'view2': [0.9, 0.11, 0.19]
    }
    result, drift_magnitude = differential_contrastive_integrity_check(data1, data2, schema, embeddings)
    print(f'Integrity Check Result: {result}, Drift Magnitude: {drift_magnitude:.4f}')