Use this Skill for research data lifecycle management: codebook generation, DVC versioning, anonymization (k-anonymity, pseudonymization), and README templates.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
Use this Skill for research data lifecycle management: codebook generation, DVC versioning, anonymization (k-anonymity, pseudonymization), and README templates.
Research Data Management โ Codebook, DVC, and Anonymization
TL;DR โ Manage the full research data lifecycle: auto-generate codebooks
from pandas DataFrames, apply k-anonymity checks and pseudonymization
(SHA-256 hashing, date shifting), version data files with DVC, define
reproducible pipelines in dvc.yaml, and produce FAIR-compliant README
documentation.
When to Use This Skill
Use this Skill whenever you need to:
Document a new dataset with a machine-readable codebook (variable names, types,
missing rates, value ranges)
Assess and enforce k-anonymity before sharing data with collaborators
Pseudonymize PII columns (IDs, names, emails, dates) before archival or transfer
Version large data files with DVC so they are tracked in Git without being
committed to the repository
Define and run a reproducible data pipeline using dvc.yaml stages
Write a FAIR-compliant README_data.md with provenance, licensing, and citation
information
Self-assess dataset compliance against FAIR principles (Findable, Accessible,
Interoperable, Reusable)
Task
When to apply
Codebook generation
After data collection; before sharing
k-anonymity check
Before external data sharing or publication
Pseudonymization
Compliance with GDPR, HIPAA, or IRB conditions
DVC data versioning
Whenever data files exceed 10 MB or must be reproduced
dvc.yaml pipeline
Multi-step analysis with cacheable intermediate outputs
README_data.md
Data deposition in Zenodo, OSF, Figshare, or institutional repos
Define variable names, units, and coding schema before collecting
Cleaning
Validate ranges, flag outliers, impute or remove missing values
Analysis
Version code and data together; cache pipeline outputs
Archival
Generate codebook, pseudonymize, deposit in repository with README
Codebook
A codebook describes every variable in a dataset. Minimum fields:
Field
Description
variable
Column name in the data file
label
Human-readable description
dtype
Python/pandas dtype
n_unique
Number of distinct non-null values
missing_pct
Percentage of missing values
min / max
Range for numeric variables
example_values
Up to 3 representative values
k-Anonymity
A dataset satisfies k-anonymity if every combination of quasi-identifier
(QI) values appears in at least k rows. Quasi-identifiers are attributes that
could be linked to external records (age, sex, ZIP code, diagnosis date).
Generalization strategies to achieve k-anonymity:
Age โ age group (e.g., 5-year bands: 20โ24, 25โ29)
ZIP code โ first 3 digits (region)
Exact date โ year-month or year only
Pseudonymization
Pseudonymization replaces direct identifiers with artificial keys. The
mapping is stored separately and protected. Direct identifiers:
Patient ID, social security number โ SHA-256 hash (deterministic, one-way)
Name, email โ hash or delete
Dates of birth / events โ date shifting (add a per-individual random offset
that preserves within-individual intervals)
SHA-256 is preferred over MD5 because it has no known practical collision
attacks and is acceptable under GDPR recital 26.
DVC โ Data Version Control
DVC tracks large data files outside Git using content-addressable storage.
Key commands:
import hashlib
import os
import numpy as np
import pandas as pd
from datetime import timedelta
from dotenv import load_dotenv
load_dotenv()
# Salt for HMAC-style hashing (never store in code โ use env var or secrets vault)# export HASH_SALT="<paste-your-salt>"
HASH_SALT: str = os.getenv("HASH_SALT", "default_salt_change_in_production")
defhash_identifier(value: str, salt: str = HASH_SALT) -> str:
"""
One-way SHA-256 hash of a string identifier with a salt.
Args:
value: Original identifier (e.g., participant ID, email).
salt: Per-study salt to prevent rainbow table attacks.
Returns:
16-character hex digest (truncated SHA-256).
"""
salted = f"{salt}:{value}"return hashlib.sha256(salted.encode("utf-8")).hexdigest()[:16]
defpseudonymize_pii_columns(
df: pd.DataFrame,
id_cols: list[str],
drop_original: bool = True,
) -> pd.DataFrame:
"""
Replace PII identifier columns with SHA-256 pseudonyms.
Args:
df: Input DataFrame containing PII.
id_cols: List of columns to pseudonymize (e.g., ['participant_id', 'email']).
drop_original: If True, drop the original columns after pseudonymization.
Returns:
DataFrame with pseudonymized ID columns (suffixed with '_pseudo').
"""
df = df.copy()
for col in id_cols:
if col notin df.columns:
continue
df[f"{col}_pseudo"] = df[col].astype(str).apply(hash_identifier)
if drop_original:
df.drop(columns=[col], inplace=True)
return df
defshift_dates(
df: pd.DataFrame,
date_cols: list[str],
id_col: str,
shift_range_days: tuple[int, int] = (-365, 365),
seed: int = 99,
) -> pd.DataFrame:
"""
Apply a per-individual random date shift to preserve within-subject intervals.
Each individual receives a single random offset (in days), applied
consistently to all their date columns. This preserves the time elapsed
between events while obscuring absolute dates.
Args:
df: DataFrame with date columns (datetime or string YYYY-MM-DD).
date_cols: List of date column names to shift.
id_col: Column identifying the individual (used for consistent offset).
shift_range_days: Tuple (min_days, max_days) for the uniform random offset.
seed: Random seed for reproducibility.
Returns:
DataFrame with shifted date columns.
"""
df = df.copy()
# Parse date columnsfor col in date_cols:
df[col] = pd.to_datetime(df[col], errors="coerce")
rng = np.random.default_rng(seed)
unique_ids = df[id_col].unique()
offset_map = {
uid: int(rng.integers(shift_range_days[0], shift_range_days[1] + 1))
for uid in unique_ids
}
for col in date_cols:
df[col] = df.apply(
lambda row: (row[col] + timedelta(days=offset_map[row[id_col]]))
if pd.notna(row[col]) else pd.NaT,
axis=1,
)
return df
defgeneralize_age(
df: pd.DataFrame,
age_col: str,
bin_width: int = 5,
output_col: str | None = None,
) -> pd.DataFrame:
"""
Replace exact age values with age-group bins (quasi-identifier generalization).
Args:
df: Input DataFrame.
age_col: Column name containing age in years (numeric).
bin_width: Width of age bins in years (default 5: 20-24, 25-29, ...).
output_col: Output column name; defaults to '{age_col}_group'.
Returns:
DataFrame with added age-group column.
"""
df = df.copy()
out_col = output_col orf"{age_col}_group"
min_age = int(df[age_col].min()) - (int(df[age_col].min()) % bin_width)
max_age = int(df[age_col].max()) + bin_width
bins = list(range(min_age, max_age + bin_width, bin_width))
labels = [f"{b}-{b + bin_width - 1}"for b in bins[:-1]]
df[out_col] = pd.cut(df[age_col], bins=bins, labels=labels, right=False)
return df
defcheck_k_anonymity(
df: pd.DataFrame,
quasi_identifiers: list[str],
k: int = 5,
) -> tuple[bool, pd.DataFrame]:
"""
Check whether a dataset satisfies k-anonymity for given quasi-identifiers.
Args:
df: DataFrame to assess.
quasi_identifiers: List of quasi-identifier column names.
k: Minimum group size required for k-anonymity.
Returns:
Tuple of (passes_k_anonymity: bool, violation_groups: DataFrame).
"""
qi_groups = df.groupby(quasi_identifiers, observed=True).size().reset_index(name="count")
violations = qi_groups[qi_groups["count"] < k]
passes = len(violations) == 0
status = "PASS"if passes elsef"FAIL โ {len(violations)} groups have count < {k}"print(f"k-anonymity check (k={k}): {status}")
ifnot passes:
print("Violating groups (sample):")
print(violations.head(10).to_string(index=False))
return passes, violations
if __name__ == "__main__":
rng = np.random.default_rng(0)
n = 100
df_sensitive = pd.DataFrame({
"participant_id": [f"PT{i:04d}"for i inrange(n)],
"name": [f"Patient_{i}"for i inrange(n)],
"email": [f"patient{i}@clinic.org"for i inrange(n)],
"age": rng.integers(20, 85, n),
"sex": rng.choice(["M", "F"], n),
"zip_code": rng.choice(["10001", "10002", "10003", "90210"], n),
"enroll_date": pd.date_range("2020-01-01", periods=n, freq="3D"),
"last_visit": pd.date_range("2023-01-01", periods=n, freq="5D"),
"diagnosis": rng.choice(["AD", "MCI", "Control"], n),
})
# Step 1: pseudonymize IDs
df_pseudo = pseudonymize_pii_columns(
df_sensitive, id_cols=["participant_id", "name", "email"]
)
# Step 2: shift dates
df_pseudo = shift_dates(
df_pseudo,
date_cols=["enroll_date", "last_visit"],
id_col="participant_id_pseudo",
shift_range_days=(-180, 180),
)
# Step 3: generalize age
df_pseudo = generalize_age(df_pseudo, age_col="age", bin_width=5)
df_pseudo = df_pseudo.drop(columns=["age"]) # drop exact age# Step 4: k-anonymity check
qi_cols = ["age_group", "sex", "zip_code"]
passes, violations = check_k_anonymity(df_pseudo, qi_cols, k=5)
df_pseudo.to_csv("pseudonymized_data.csv", index=False)
print(f"\nPseudonymized dataset shape: {df_pseudo.shape}")
print(df_pseudo.head(5).to_string(index=False))
Step 3 โ DVC Pipeline YAML and dvc repro
import os
import subprocess
import textwrap
from pathlib import Path
from dotenv import load_dotenv
load_dotenv()
DVC_YAML_CONTENT = textwrap.dedent("""
stages:
preprocess:
cmd: python scripts/preprocess.py
deps:
- data/raw/cohort_data.csv
- scripts/preprocess.py
params:
- params.yaml:
- preprocess.missing_threshold
- preprocess.outlier_z_threshold
outs:
- data/processed/cohort_clean.csv
train:
cmd: python scripts/train.py
deps:
- data/processed/cohort_clean.csv
- scripts/train.py
params:
- params.yaml:
- train.model_type
- train.n_estimators
- train.max_depth
- train.random_state
outs:
- models/model.pkl
metrics:
- results/metrics.json:
cache: false
evaluate:
cmd: python scripts/evaluate.py
deps:
- models/model.pkl
- data/processed/cohort_clean.csv
- scripts/evaluate.py
outs:
- results/predictions.csv
plots:
- results/roc_curve.csv:
cache: false
- results/shap_summary.png:
cache: false
""").strip()
PARAMS_YAML_CONTENT = textwrap.dedent("""
preprocess:
missing_threshold: 0.20 # drop columns with > 20% missing
outlier_z_threshold: 3.0 # z-score threshold for outlier flagging
train:
model_type: random_forest
n_estimators: 200
max_depth: 10
random_state: 42
evaluate:
threshold: 0.5 # decision threshold for binary classification
""").strip()
defwrite_dvc_pipeline(
project_root: str = ".",
dvc_yaml_path: str = "dvc.yaml",
params_yaml_path: str = "params.yaml",
) -> None:
"""
Write dvc.yaml and params.yaml to the project root.
Args:
project_root: Root directory of the DVC-initialized project.
dvc_yaml_path: Relative path for the DVC pipeline file.
params_yaml_path: Relative path for the parameters file.
"""
root = Path(project_root)
dvc_path = root / dvc_yaml_path
params_path = root / params_yaml_path
withopen(dvc_path, "w", encoding="utf-8") as fh:
fh.write(DVC_YAML_CONTENT + "\n")
print(f"Written: {dvc_path}")
withopen(params_path, "w", encoding="utf-8") as fh:
fh.write(PARAMS_YAML_CONTENT + "\n")
print(f"Written: {params_path}")
defrun_dvc_command(args: list[str], cwd: str = ".") -> int:
"""
Run a DVC CLI command as a subprocess.
Args:
args: DVC subcommand and arguments (e.g., ['dvc', 'repro']).
cwd: Working directory to run the command in.
Returns:
Return code (0 = success).
"""
result = subprocess.run(args, cwd=cwd, capture_output=True, text=True)
if result.stdout:
print(result.stdout)
if result.stderr:
print(result.stderr)
return result.returncode
defsetup_dvc_remote(
remote_name: str = "myremote",
remote_url: str | None = None,
) -> None:
"""
Configure a DVC remote storage backend.
Args:
remote_name: Alias for the remote (e.g., 'myremote', 's3remote').
remote_url: Storage URL. Reads DVC_REMOTE_URL env var if not provided.
Examples:
Local: /data/dvc-store
S3: s3://my-bucket/dvc-store
GCS: gs://my-bucket/dvc-store
"""
url = remote_url or os.getenv("DVC_REMOTE_URL", "/tmp/dvc-store")
run_dvc_command(["dvc", "remote", "add", "--default", remote_name, url])
print(f"DVC remote '{remote_name}' configured at: {url}")
if __name__ == "__main__":
# In practice, run from inside the Git+DVC-initialized project directory
write_dvc_pipeline()
print("\nTo run the pipeline:")
print(" dvc repro")
print(" dvc push")
print("\nTo track a data file:")
print(" dvc add data/raw/cohort_data.csv")
print(" git add data/raw/cohort_data.csv.dvc .gitignore")
print(' git commit -m "Track raw data with DVC"')
Advanced Usage
FAIR Assessment Checklist
FAIR_CHECKLIST = {
"Findable": [
"Dataset has a globally unique persistent identifier (DOI via Zenodo/Figshare)",
"Metadata is registered in a searchable resource (DataCite, Google Dataset Search)",
"Metadata includes dataset title, author, creation date, keywords",
"Identifier is included in the metadata and in the data file itself",
],
"Accessible": [
"Dataset is retrievable via standard open protocol (HTTPS)",
"Protocol is free and open (no proprietary software required)",
"Access conditions are clearly stated (open / restricted / embargoed)",
"Metadata remains accessible even if dataset is unavailable",
],
"Interoperable": [
"Data use a formal, accessible, shared, and broadly applicable language (CSV/JSON/NetCDF)",
"Data use FAIR-compliant vocabularies and ontologies (MeSH, SNOMED, OBI)",
"Data include qualified references to other (meta)data",
],
"Reusable": [
"Data are released with a clear and accessible data usage license (CC-BY or CC0)",
"Data are associated with detailed provenance (collection methods, instruments)",
"Data meet domain-relevant community standards (MIAME, CONSORT, CDISC)",
"README includes citation information and contact for reuse requests",
],
}
defassess_fair_compliance(checklist: dict[str, list[str]]) -> dict:
"""
Interactively (or programmatically) assess FAIR compliance.
For automated use, pass pre-filled responses instead of prompting.
Here we print the checklist for manual self-assessment.
"""
scores = {}
for principle, criteria in checklist.items():
print(f"\n{'='*60}")
print(f" {principle}")
print(f"{'='*60}")
for i, criterion inenumerate(criteria, 1):
print(f" {i}. {criterion}")
scores[principle] = len(criteria)
total = sum(scores.values())
print(f"\nTotal checklist items: {total}")
print("Review each item and record Y/N in your DMP or README.")
return scores
# Run the FAIR assessment
assess_fair_compliance(FAIR_CHECKLIST)
README_data.md Template
README_DATA_TEMPLATE = """
# Dataset: {dataset_title}
## Description
{description}
## Variables
See `codebook.csv` for a full variable dictionary including type, range,
missing rates, and example values.
## Provenance
- **Source**: {source}
- **Collection period**: {collection_period}
- **Collection method**: {collection_method}
- **IRB / Ethics approval**: {irb_number}
## File structure
data/
raw/ # Original unmodified data files (tracked with DVC)
processed/ # Cleaned and pseudonymized versions
codebook.csv # Auto-generated variable dictionary
README_data.md # This file
dvc.yaml # Pipeline definition
params.yaml # Analysis parameters
## License
{license}
## Citation
If you use this dataset, please cite:
{citation}
## Contact
{contact_email}
"""
def write_readme_data(
output_path: str = "README_data.md",
**fields,
) -> None:
"""Write a populated README_data.md using the standard template."""
content = README_DATA_TEMPLATE.format(**{
"dataset_title": fields.get("dataset_title", "Research Dataset"),
"description": fields.get("description", ""),
"source": fields.get("source", ""),
"collection_period": fields.get("collection_period", ""),
"collection_method": fields.get("collection_method", ""),
"irb_number": fields.get("irb_number", ""),
"license": fields.get("license", "CC-BY 4.0"),
"citation": fields.get("citation", "Author et al. (Year). Title. DOI."),
"contact_email": fields.get("contact_email", ""),
})
with open(output_path, "w", encoding="utf-8") as fh:
fh.write(content)
print(f"README_data.md written to {output_path}")
Troubleshooting
Problem
Likely cause
Fix
dvc repro shows no changes
No dependency has changed
Force re-run with dvc repro --force
dvc push authentication error
Missing cloud credentials
Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY env vars
k-anonymity violations after generalization
Rare combination of QIs
Increase bin width; suppress or merge rare groups
SHA-256 collision risk
Not a practical concern
SHA-256 has no known practical collisions; safe for pseudonymization
Date shift changes interval lengths
Shifting by different offsets per visit
Use per-individual (not per-row) offset โ assign once per id_col
Codebook shows wrong dtype
Mixed-type column
Cast explicitly before calling generate_codebook()
Example 3 โ DVC Pipeline yaml + dvc repro Walkthrough
from pathlib import Path
# Write pipeline definition files
write_dvc_pipeline(project_root=".", dvc_yaml_path="dvc.yaml",
params_yaml_path="params.yaml")
# Typical DVC workflow after writing the pipeline:print("""
# Add a raw data file to DVC tracking
dvc add data/raw/cohort_data.csv
# Commit the .dvc tracking file and .gitignore
git add data/raw/cohort_data.csv.dvc data/.gitignore dvc.yaml params.yaml
git commit -m "Add data pipeline and track raw data"
# Configure remote and push data
dvc remote add -d myremote /data/shared/dvc-store
dvc push
# Run the full pipeline
dvc repro
# Check what changed
dvc status
dvc diff
# Run an experiment with a different parameter
dvc exp run --set-param train.n_estimators=300
# Show all experiments
dvc exp show
""")
