在即時政策限制下保障私隱的衞星異常應對主動學習

Privacy-Preserving Active Learning for satellite anomaly response operations under real-time policy constraints

Introduction: A Lesson from a Silent Satellite

Last year, while working on a small cubesat telemetry pipeline for a university research group, I hit a wall that completely reshaped how I think about machine learning in space systems. We had trained a solid anomaly detector on historical telemetry, but the moment we tried to deploy it operationally, two problems collided: we couldn't ship raw telemetry to a central training server (privacy and export-control concerns), and our operators needed decisions now — not after a nightly batch retrain. The satellite was already doing something weird, and our "smart" model was confidently wrong.

While exploring solutions to this, I discovered that the intersection of active learning, differential privacy, and real-time policy enforcement is one of the most underexplored and genuinely hard problems in operational AI. In my research of federated learning papers, I realized most assume generous latency budgets and clean communication channels — neither of which exist in a satellite ground-segment context. Through studying policy-as-code frameworks and experimenting with constrained optimization, I built a small prototype that I want to share here.

This article walks through the architecture I landed on: a privacy-preserving active learning loop for satellite anomaly response that respects hard, real-time policy constraints (e.g., "never uplink a command that could disable the ADCS during an eclipse"). Along the way I'll share code patterns, the traps I fell into, and where I think this technology is heading.

Why This Problem Is Harder Than It Looks

Satellite anomaly response sits at a nasty intersection:

• Data is sensitive: Telemetry may contain operational details, proprietary instrument readings, or information governed by export regulations (ITAR in the US, for example). You often cannot centralize it.
• Labels are expensive: An anomaly label requires an operator's attention, and operators are a scarce, expensive resource during a pass window.
• Latency is brutal: Ground-station contact windows can be minutes long. A policy decision (e.g., "this maneuver violates thermal constraints") must be made before the command is uplinked.
• Policies change: A constraint that holds during nominal ops may be overridden during a safe-mode recovery. The ML system must respect a policy that's evaluated at decision time, not training time.

Active learning is the natural fit for the labeling problem — you query the operator only for the most informative samples. But naive active learning leaks: sending a raw telemetry window to a labeling service is a privacy violation. And standard active learning has no notion of "the policy says you can't ask about this sample right now."

Technical Background: The Three Pillars

1. Active Learning Under Budget

The classic uncertainty-sampling loop picks the sample with the highest predictive entropy to label next. In our setting, the "labeling oracle" is a human operator, and each query costs precious contact time. A practical variant uses expected model change or margin sampling, both of which I found more stable than raw entropy when the anomaly class is rare.

import numpy as np
from scipy.stats import entropy

def margin_score(probs):
    """Smaller margin = more uncertain = more valuable to label."""
    p_sorted = np.sort(probs, axis=1)
    return p_sorted[:, -1] - p_sorted[:, -2]

def select_queries(model, pool_X, k, policy_mask):
    probs = model.predict_proba(pool_X)
    scores = margin_score(probs)
    # policy_mask: boolean array, False = forbidden to query
    eligible = np.where(policy_mask)[0]
    ranked = eligible[np.argsort(scores[eligible])]
    return ranked[:k]

The policy_mask is the key addition — it's where real-time constraints enter the loop.

2. Differential Privacy for Telemetry

To prevent the labeling service (or a shared model) from memorizing individual telemetry windows, I applied DP-SGD with a per-example gradient clipping and Gaussian noise mechanism. The intuition: each sample's contribution to the gradient is bounded, so no single window can dominate the model.

import torch

def dp_sgd_step(model, batch, optimizer, clip_norm=1.0, noise_mult=1.1):
    optimizer.zero_grad()
    per_example_grads = []
    for x, y in batch:
        loss = model.loss(x.unsqueeze(0), y.unsqueeze(0))
        loss.backward(retain_graph=True)
        grads = [p.grad.clone() for p in model.parameters()]
        per_example_grads.append(grads)
        model.zero_grad()

    # Clip each per-example gradient
    clipped = []
    for grads in per_example_grads:
        norm = torch.sqrt(sum((g**2).sum() for g in grads))
        scale = min(1.0, clip_norm / (norm + 1e-6))
        clipped.append([g * scale for g in grads])

    # Sum, add Gaussian noise, apply
    for i, p in enumerate(model.parameters()):
        summed = sum(c[i] for c in clipped)
        noise = torch.randn_like(summed) * noise_mult * clip_norm
        p.grad = (summed + noise) / len(batch)
    optimizer.step()

One interesting finding from my experimentation with DP-SGD on telemetry: temporal correlation breaks the privacy guarantee's assumptions. Consecutive samples in a window are highly correlated, so naive per-sample clipping under-protects. I ended up grouping by "session" and clipping at the session level — a subtle but important fix.

3. Real-Time Policy Constraints

The policy layer is where this gets operationally interesting. Constraints like "don't uplink attitude commands during eclipse" or "anomaly response must not exceed ΔV budget" are naturally expressed as linear inequalities over the action space. I modeled the decision as a constrained optimization:

from scipy.optimize import minimize

def constrained_response(model_probs, action_matrix, constraints, budget):
    """
    Choose an action that maximizes expected anomaly mitigation
    subject to A @ x <= b and sum(x) <= budget.
    """
    A, b = constraints  # policy-derived
    n_actions = action_matrix.shape[0]

    def neg_utility(x):
        return -np.dot(model_probs, action_matrix @ x)

    cons = [{'type': 'ineq', 'fun': lambda x: b - A @ x},
            {'type': 'ineq', 'fun': lambda x: budget - x.sum()}]
    bounds = [(0, 1)] * n_actions
    res = minimize(neg_utility, np.zeros(n_actions),
                   method='SLSQP', constraints=cons, bounds=bounds)
    return res.x

The policy constraints are evaluated at inference time, so a mid-pass policy update (e.g., ground control flips the system into safe mode) immediately changes the feasible action set. This is the "real-time" part: the model's recommendation is always filtered through a live policy check.

Implementation: The Integrated Loop

Here's the architecture that emerged from my experimentation. The pipeline runs in two phases per contact window:

1. Onboard: A lightweight model scores telemetry locally and selects a small batch of candidate queries. Only these candidates (with DP noise added to any statistics) leave the spacecraft.
2. Ground: The operator labels the candidates, the model updates via DP-SGD, and a policy-checked action recommendation is uplinked.

class PrivacyPreservingActiveLearner:
    def __init__(self, model, policy_engine, epsilon_budget):
        self.model = model
        self.policy = policy_engine
        self.eps = epsilon_budget
        self.round = 0

    def onboard_select(self, telemetry_pool, k=5):
        # Local scoring only; nothing leaves except indices + DP stats
        mask = self.policy.query_allowed(telemetry_pool)
        idx = select_queries(self.model, telemetry_pool, k, mask)
        return idx  # only indices leave the spacecraft

    def ground_update(self, labeled_batch):
        # Check privacy budget before spending it
        cost = self.privacy_accountant.cost(len(labeled_batch))
        if self.eps - cost < 0:
            raise RuntimeError("Privacy budget exhausted")
        dp_sgd_step(self.model, labeled_batch, self.optimizer)
        self.eps -= cost
        self.round += 1
        return self.model

    def decide(self, state, policy_snapshot):
        probs = self.model.predict_proba(state)
        return constrained_response(
            probs, policy_snapshot.action_matrix, policy_snapshot.constraints,
            policy_snapshot.budget)

The policy_engine.query_allowed call is the piece I underestimated at first. In my early prototype, I only enforced policy on actions, not on queries. That was a mistake: querying an operator about a classified instrument's telemetry is itself a policy violation. Once I moved the policy check to gate the query selection, the system became genuinely deployable.

Real-World Applications

While learning about operational satellite fleets, I observed that this pattern generalizes well beyond space:

• Autonomous drones: The same privacy-preserving active learning loop applies to swarms where each drone's sensor data is sensitive, but the fleet benefits from shared anomaly models.
• Medical monitoring: Wearable telemetry has identical properties — sensitive, expensive labels, and regulatory constraints (HIPAA) that act exactly like real-time policies.
• Industrial IoT: A factory floor where raw sensor streams can't leave the premises but a shared anomaly model is valuable.

The unifying theme is federated, budgeted, constrained learning — and I suspect this will be one of the defining patterns of agentic AI in regulated environments.

Challenges and Solutions

Challenge 1: The privacy budget is a real-time constraint too. I initially treated ε as a static hyperparameter. But in a live system, you're spending budget continuously, and the accountant must run inside the decision loop. My solution was to expose the remaining budget as a first-class input to the policy engine, so the system can fall back to a non-learning heuristic when the budget runs low.

Challenge 2: Policy latency. Evaluating SLSQP constraints on a radiation-hardened flight computer is not free. I found that precompiling the constraint matrix and using a fast interior-point solver (or even a lookup table for small action spaces) cut decision latency from ~40ms to under 2ms — critical when your contact window is 90 seconds.

Challenge 3: Distribution shift during anomalies. Active learning assumes the pool is representative. During a genuine anomaly, it isn't — the very samples you most want to label are the ones your model has never seen. My exploration of this revealed that a small amount of exploration (random sampling with probability ε_explore) alongside uncertainty sampling dramatically improved recovery from novel anomalies.

def epsilon_explore(eligible, scores, k, eps_explore=0.1):
    n_explore = int(k * eps_explore)
    explore = np.random.choice(eligible, n_explore, replace=False)
    exploit = eligible[np.argsort(scores[eligible])][:k - n_explore]
    return np.concatenate([explore, exploit])

Future Directions

As I was experimenting with this stack, I kept running into the same question: can we push more of the learning onboard? Radiation-tolerant FPGAs and emerging edge accelerators make local DP-SGD feasible, which would eliminate the round-trip entirely. I'm also watching the intersection with quantum-assisted optimization — constrained policy optimization is a natural fit for QAOA-style solvers, and while current hardware isn't there yet, the formulation (quadratic objectives with linear constraints) maps cleanly onto annealing hardware. That's a research direction I plan to explore next.

Another thread: agentic policy negotiation. Instead of a static policy engine, imagine an agent that reasons about policy conflicts (e.g., a thermal constraint vs. an attitude constraint) and proposes a resolution for human approval. That's where active learning meets agentic AI, and it's genuinely exciting.

Conclusion: What I Learned

My exploration of this problem taught me three things I'll carry into every future system I build:

1. Privacy and latency are not afterthoughts — they're architectural. The moment I treated the privacy budget and the policy engine as first-class citizens in the decision loop, the design simplified dramatically.
2. Active learning is a policy problem, not just a statistics problem. The query itself is an action, and actions have constraints.
3. Constrained optimization at inference time is underrated. Filtering model outputs through a live, fast policy check is a simple pattern that makes ML deployable in regulated environments.

If you're building any system where data is sensitive and decisions are time-critical, the privacy-preserving active learning loop with real-time policy enforcement is a pattern worth having in your toolkit. I'd love to hear how others have tackled the temporal-correlation privacy issue — that's the piece I'm still chewing on.