发现数据，提交研究，取回结果。

import os
import httpx

# Install httpx; load the key from the environment, never from committed source.
with httpx.Client(base_url="https://datapanel.dev", timeout=30, follow_redirects=False) as client:
    headers = {"X-API-Key": os.environ["DATAPANEL_API_KEY"]}
    for path in ("/v1/me", "/v1/compute/profiles"):
        response = client.get(path, headers=headers)
        # Redirects and error bodies must not be used to disclose credentials.
        if response.status_code != 200:
            raise RuntimeError(f"Request failed: HTTP {response.status_code}")
        print(path, "available")
# Continue with the public SDK for quotes, persisted submissions and private exports.

# Set DATAPANEL_API_KEY through your secret manager or shell environment.
# Read current account and capabilities. Never add --location to authenticated requests.
curl --fail --silent --show-error -H "X-API-Key: $DATAPANEL_API_KEY" https://datapanel.dev/v1/me
curl --fail --silent --show-error -H "X-API-Key: $DATAPANEL_API_KEY" https://datapanel.dev/v1/compute/profiles
# Browse available partitions; historical market-file downloads are unavailable.
curl --fail --silent --show-error -H "X-API-Key: $DATAPANEL_API_KEY" 'https://datapanel.dev/v1/catalog?limit=20'
# Use the SDK to freeze inputs, quote within a budget and persist submission state.

五类策略 Demo

趋势、均值回归、突破、截面排序和波动率缩放示例。仅演示因果数据接口，未验证盈利能力；不自动提交执行。

"""Causal signal examples, not validated investment strategies.

Input CSV: symbol,timestamp_ms,available_at_ms,close.
`timestamp_ms` is the completed bar's timestamp. `available_at_ms` is when its
close actually became observable (UTC milliseconds). No future price is used.
Output: symbol,timestamp_ms,available_at_ms,prediction (target_fraction).
Execute signals no earlier than availability, normally next tradable bar.
"""

import argparse
import csv
import itertools
import math
from collections import defaultdict, deque


def signals(rows, strategy="trend", window=20):
    if window < 2:
        raise ValueError("window must be >=2")
    histories = defaultdict(lambda: deque(maxlen=window))
    previous = -1
    for stamp, group in itertools.groupby(rows, key=lambda row: int(row["timestamp_ms"])):
        if stamp <= previous:
            raise ValueError("Rows must be grouped and strictly ordered by timestamp")
        previous = stamp
        batch = list(group)
        seen = set()
        values = []
        for row in batch:
            symbol = row["symbol"]
            price = float(row["close"])
            available = int(row["available_at_ms"])
            if symbol in seen or available < stamp or not math.isfinite(price) or price <= 0:
                raise ValueError("Duplicate symbol, invalid availability or invalid close")
            seen.add(symbol)
            history = histories[symbol]
            score = 0.0
            if len(history) == window:
                if strategy in ("trend", "cross_sectional"):
                    score = price / history[0] - 1
                elif strategy == "mean_reversion":
                    mean = sum(history) / len(history)
                    score = mean / price - 1
                elif strategy == "breakout":
                    score = 1.0 if price > max(history) else -1.0 if price < min(history) else 0.0
                elif strategy == "volatility_scaled":
                    returns = [history[i] / history[i - 1] - 1 for i in range(1, len(history))]
                    volatility = math.sqrt(sum(v * v for v in returns) / len(returns))
                    score = (price / history[0] - 1) / max(volatility * math.sqrt(window), 1e-6)
                else:
                    raise ValueError("Unknown strategy")
            history.append(price)
            values.append([symbol, stamp, available, score])
        if strategy == "cross_sectional":
            # All selected cross-sectional inputs must be available before any decision.
            ready = max(v[2] for v in values)
            scores = sorted(set(v[3] for v in values))
            for v in values:
                v[2] = ready
                v[3] = (
                    0.0
                    if len(scores) < 2
                    else (2 * scores.index(v[3]) / (len(scores) - 1) - 1) / len(values)
                )
        for symbol, stamp, available, score in values:
            yield {
                "symbol": symbol,
                "timestamp_ms": stamp,
                "available_at_ms": available,
                "prediction": max(-1.0, min(1.0, score)),
            }


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input")
    parser.add_argument("output")
    parser.add_argument(
        "--strategy",
        choices=["trend", "mean_reversion", "breakout", "cross_sectional", "volatility_scaled"],
        default="trend",
    )
    parser.add_argument("--window", type=int, default=20)
    args = parser.parse_args()
    with open(args.input, newline="") as source, open(args.output, "w", newline="") as target:
        writer = csv.DictWriter(
            target, fieldnames=["symbol", "timestamp_ms", "available_at_ms", "prediction"]
        )
        writer.writeheader()
        writer.writerows(signals(csv.DictReader(source), args.strategy, args.window))

算力报价 Demo（不执行）

"""Save owned source, resolve data and request a quote. Does NOT execute code.

Set DATAPANEL_API_KEY. Choose a selection actually present in /v1/catalog.
The matching source must obey the compute runtime contract before execution is opened.
"""

from pathlib import Path

from datapanel_sdk import Datapanel


def plan(client: Datapanel, source_path, selection):
    source = client.request(
        "POST",
        "/v1/compute/sources",
        {"filename": Path(source_path).name, "content": Path(source_path).read_text()},
    )
    snapshot = client.request("POST", "/v1/compute/data-selections", selection)
    quote = client.request(
        "POST",
        "/v1/compute/quotes",
        {
            "profile": "cpu-small",
            "source_artifact_id": source["id"],
            "dataset_snapshot_id": snapshot["id"],
            "max_wall_seconds": 1800,
            "max_credits": "2",
        },
    )
    return {"source": source, "snapshot": snapshot, "quote": quote, "execution_enabled": False}