brycewang-stanford Data Collection Automation Guide
A skill for automating research data collection, survey deployment, and data pipeline management. Covers survey platform APIs, automated data retrieval, quality checks, ETL pipelines, and scheduling for longitudinal studies.
Survey Platform APIs
Qualtrics API
import os
import json
import urllib.request
import time
def export_qualtrics_responses(survey_id: str,
file_format: str = "csv") -> str:
"""
Export survey responses from Qualtrics via API.
Args:
survey_id: The Qualtrics survey ID (SV_...)
file_format: Export format (csv, json, spss)
"""
api_token = os.environ["QUALTRICS_API_TOKEN"]
data_center = os.environ["QUALTRICS_DATACENTER"]
base_url = f"https://{data_center}.qualtrics.com/API/v3"
headers = {
"X-API-TOKEN": api_token,
"Content-Type": "application/json"
}
# Step 1: Start export
export_data = json.dumps({
"format": file_format,
"compress": False
}).encode("utf-8")
req = urllib.request.Request(
f"{base_url}/surveys/{survey_id}/export-responses",
data=export_data,
headers=headers
)
response = json.loads(urllib.request.urlopen(req).read())
progress_id = response["result"]["progressId"]
# Step 2: Poll for completion
status = "inProgress"
while status == "inProgress":
time.sleep(2)
req = urllib.request.Request(
f"{base_url}/surveys/{survey_id}/export-responses/{progress_id}",
headers=headers
)
check = json.loads(urllib.request.urlopen(req).read())
status = check["result"]["status"]
file_id = check["result"]["fileId"]
# Step 3: Download file
req = urllib.request.Request(
f"{base_url}/surveys/{survey_id}/export-responses/{file_id}/file",
headers=headers
)
file_data = urllib.request.urlopen(req).read()
output_path = f"responses_{survey_id}.{file_format}"
with open(output_path, "wb") as f:
f.write(file_data)
return output_path
REDCap API
def export_redcap_records(api_url: str, fields: list[str] = None) -> list:
"""
Export records from a REDCap project.
Args:
api_url: REDCap API endpoint URL
fields: List of field names to export (None = all fields)
"""
api_token = os.environ["REDCAP_API_TOKEN"]
data = {
"token": api_token,
"content": "record",
"format": "json",
"type": "flat"
}
if fields:
data["fields"] = ",".join(fields)
encoded = urllib.parse.urlencode(data).encode("utf-8")
req = urllib.request.Request(api_url, data=encoded)
response = urllib.request.urlopen(req)
return json.loads(response.read())
Automated Data Quality Checks
Validation Pipeline
import pandas as pd
from datetime import datetime
def validate_survey_data(df: pd.DataFrame,
rules: dict) -> dict:
"""
Run automated data quality checks on collected data.
Args:
df: DataFrame of survey responses
rules: Dict of column -> validation rule pairs
"""
issues = []
# Check for duplicates
dupes = df.duplicated(subset=["respondent_id"]).sum()
if dupes > 0:
issues.append(f"Found {dupes} duplicate respondent IDs")
# Check completion rates
completion = df.notna().mean()
low_completion = completion[completion < 0.5]
for col in low_completion.index:
issues.append(f"Column '{col}' has {low_completion[col]:.0%} completion")
# Check value ranges
for col, rule in rules.items():
if col not in df.columns:
continue
if "min" in rule:
violations = (df[col] < rule["min"]).sum()
if violations > 0:
issues.append(f"{violations} values below minimum in '{col}'")
if "max" in rule:
violations = (df[col] > rule["max"]).sum()
if violations > 0:
issues.append(f"{violations} values above maximum in '{col}'")
# Check for speeding (unusually fast completion)
if "duration_seconds" in df.columns:
median_time = df["duration_seconds"].median()
speeders = (df["duration_seconds"] < median_time * 0.3).sum()
if speeders > 0:
issues.append(f"{speeders} respondents completed in <30% of median time")
return {
"n_records": len(df),
"n_issues": len(issues),
"issues": issues,
"timestamp": datetime.now().isoformat()
}
ETL Pipeline for Research Data
Scheduled Data Retrieval
def research_etl_pipeline(sources: list[dict],
output_dir: str) -> dict:
"""
Extract, transform, and load research data from multiple sources.
Args:
sources: List of data source configurations
output_dir: Directory to save processed data
"""
results = {}
for source in sources:
name = source["name"]
# Extract
if source["type"] == "qualtrics":
raw_path = export_qualtrics_responses(source["survey_id"])
df = pd.read_csv(raw_path)
elif source["type"] == "redcap":
records = export_redcap_records(source["api_url"])
df = pd.DataFrame(records)
elif source["type"] == "csv_url":
df = pd.read_csv(source["url"])
else:
continue
# Transform
df = df.dropna(how="all")
df.columns = [c.strip().lower().replace(" ", "_") for c in df.columns]
# Load
timestamp = datetime.now().strftime("%Y%m%d")
output_path = f"{output_dir}/{name}_{timestamp}.csv"
df.to_csv(output_path, index=False)
results[name] = {
"records": len(df),
"columns": len(df.columns),
"output": output_path
}
return results
Scheduling and Monitoring
Cron-Based Scheduling
# Run data collection pipeline daily at 6 AM
# crontab -e
0 6 * * * cd /path/to/project && python collect_data.py >> logs/collection.log 2>&1
Monitoring Checklist
For longitudinal studies, automate monitoring of:
- Response rates per wave (alert if below threshold)
- Data quality metrics (completion, speeding, straight-lining)
- API quota usage (stay within rate limits)
- Storage usage and backup status
- Participant dropout patterns
Ethical Considerations
Always ensure automated data collection complies with your IRB/ethics board approval. Store API tokens securely using environment variables, never in code. Implement data encryption at rest. Log all data access for audit trails. Respect rate limits on external APIs. Include automated checks for consent status before processing participant data.