Code Dataset Curation: Deduplication, License Filtering, and Quality Scoring for LLM Training

Code LLMs are trained on code scraped from public repositories. The quality of the training data directly determines the quality of the model’s code generation. Feed the model duplicated, unlicensed, buggy, or secret-containing code, and it will reproduce all of those problems at inference time. The curation pipeline that transforms billions of raw source files into a clean, deduplicated, licensed, high-quality training dataset is the single most important engineering effort in building a code LLM.

The Stack v2, the largest openly documented code dataset, starts with 3.3 billion files across 619 programming languages scraped from GitHub, GitLab, and Bitbucket. After curation, roughly 800 million files remain. This post documents the full pipeline: source collection, license filtering (which drops 40-60% of files), exact and near-deduplication (another 30-50% reduction), quality scoring, language distribution balancing, and PII removal. Every stage includes implementation code.

---

1. Source Collection: The Raw Data

1.1 Scale of Public Code

GitHub alone hosts over 420 million repositories. Most are forks, empty, or trivially small. The raw numbers:

1.2 Collection Pipeline

The collection pipeline uses the GitHub API (or GH Archive for historical data) and git clone for content. The critical decisions at this stage:

import subprocess
import hashlib
from pathlib import Path

# Configuration for collection
MAX_FILE_SIZE_BYTES = 1_000_000  # 1 MB max file size
MAX_REPO_SIZE_MB = 500           # Skip repos larger than 500 MB
EXCLUDED_EXTENSIONS = {
    '.min.js', '.min.css',       # Minified files
    '.lock', '.sum',              # Lock files
    '.pb.go', '.pb.cc',          # Protobuf generated
    '.generated.ts',             # Generated code
    '.pyc', '.class', '.o',      # Compiled artifacts
    '.png', '.jpg', '.gif',      # Binary assets
    '.woff', '.ttf', '.eot',     # Fonts
    '.zip', '.tar', '.gz',       # Archives
}

def should_include_file(file_path, file_size):
    """
    First-pass filter: exclude obviously unwanted files.
    Applied during collection to reduce storage costs.
    """
    path = Path(file_path)

    # Size filter
    if file_size > MAX_FILE_SIZE_BYTES:
        return False

    # Extension filter
    if path.suffix.lower() in EXCLUDED_EXTENSIONS:
        return False

    # Path-based filters
    path_str = str(path).lower()
    excluded_dirs = [
        'node_modules/', 'vendor/', '.git/', '__pycache__/',
        'build/', 'dist/', '.tox/', '.eggs/', 'target/',
        'venv/', '.venv/', 'env/',
    ]
    for excluded in excluded_dirs:
        if excluded in path_str:
            return False

    # Skip dotfiles (configs, not source code)
    if path.name.startswith('.') and path.suffix not in {'.py', '.js', '.ts', '.rs', '.go'}:
        return False

    return True

1.3 Language Detection

File extension is a reasonable first-pass language classifier, but it is ambiguous (.h could be C or C++, .pl could be Perl or Prolog). Production pipelines use linguist (GitHub’s own tool) or guesslang:

# Using file extension mapping (simplified)
EXTENSION_TO_LANGUAGE = {
    '.py': 'Python', '.js': 'JavaScript', '.ts': 'TypeScript',
    '.java': 'Java', '.c': 'C', '.cpp': 'C++', '.cc': 'C++',
    '.h': 'C/C++', '.hpp': 'C++', '.cs': 'C#', '.go': 'Go',
    '.rs': 'Rust', '.rb': 'Ruby', '.php': 'PHP', '.swift': 'Swift',
    '.kt': 'Kotlin', '.scala': 'Scala', '.lua': 'Lua',
    '.sh': 'Shell', '.bash': 'Shell', '.zsh': 'Shell',
    '.sql': 'SQL', '.r': 'R', '.R': 'R', '.jl': 'Julia',
    '.hs': 'Haskell', '.ml': 'OCaml', '.ex': 'Elixir',
    '.clj': 'Clojure', '.lisp': 'Lisp', '.el': 'Emacs Lisp',
    '.vim': 'Vim Script', '.cu': 'CUDA', '.cuh': 'CUDA',
}

def detect_language(file_path, content=None):
    """Detect programming language from file path and content."""
    ext = Path(file_path).suffix.lower()

    # Extension-based detection
    if ext in EXTENSION_TO_LANGUAGE:
        lang = EXTENSION_TO_LANGUAGE[ext]
        # Disambiguate .h files
        if lang == 'C/C++' and content:
            if any(kw in content for kw in ['class ', 'namespace ', 'template<', 'std::']):
                return 'C++'
            return 'C'
        return lang

    # Shebang detection for extensionless scripts
    if content and content.startswith('#!'):
        first_line = content.split('\n')[0].lower()
        if 'python' in first_line:
            return 'Python'
        if 'node' in first_line or 'deno' in first_line:
            return 'JavaScript'
        if 'bash' in first_line or 'sh' in first_line:
            return 'Shell'
        if 'ruby' in first_line:
            return 'Ruby'

    return 'Unknown'

---

2. License Filtering

2.1 Why License Filtering Matters

Using code with non-permissive licenses (GPL, AGPL, SSPL) in training data creates legal risk: the model might reproduce copyleft-licensed code, potentially requiring the model’s weights or outputs to be released under the same license. The safe approach is to train only on permissive-licensed code.

Permissive licenses (MIT, Apache-2.0, BSD-2-Clause, BSD-3-Clause, ISC, Unlicense, CC0-1.0) allow use with minimal restrictions. Copyleft licenses (GPL-2.0, GPL-3.0, AGPL-3.0, LGPL-2.1, MPL-2.0) require derivative works to carry the same license.

2.2 License Detection Strategy

License detection operates at three levels:

1. Repository-level: Check for a LICENSE or COPYING file in the repo root.
2. File-level SPDX headers: Check the first 5 lines for SPDX-License-Identifier: headers.
3. License file content matching: Match the LICENSE file content against known license texts.

import re

# Permissive licenses (safe for training)
PERMISSIVE_SPDX = {
    'MIT', 'Apache-2.0', 'BSD-2-Clause', 'BSD-3-Clause',
    'ISC', 'Unlicense', 'CC0-1.0', 'BSL-1.0', 'Zlib',
    'PostgreSQL', '0BSD', 'MIT-0',
}

# Copyleft licenses (exclude from training)
COPYLEFT_SPDX = {
    'GPL-2.0-only', 'GPL-2.0-or-later', 'GPL-3.0-only', 'GPL-3.0-or-later',
    'AGPL-3.0-only', 'AGPL-3.0-or-later', 'LGPL-2.1-only', 'LGPL-2.1-or-later',
    'LGPL-3.0-only', 'LGPL-3.0-or-later', 'MPL-2.0', 'EUPL-1.2',
    'SSPL-1.0', 'CPAL-1.0',
}

# Common SPDX variants (normalize before lookup)
SPDX_ALIASES = {
    'GPL-2.0': 'GPL-2.0-only',
    'GPL-3.0': 'GPL-3.0-only',
    'AGPL-3.0': 'AGPL-3.0-only',
    'LGPL-2.1': 'LGPL-2.1-only',
    'LGPL-3.0': 'LGPL-3.0-only',
}

def extract_spdx_from_header(content):
    """
    Extract SPDX license identifier from file header.
    Checks first 5 lines for SPDX-License-Identifier.
    """
    lines = content.split('\n')[:5]
    for line in lines:
        match = re.search(r'SPDX-License-Identifier:\s*(.+)', line)
        if match:
            spdx_id = match.group(1).strip()
            # Handle compound expressions like "MIT OR Apache-2.0"
            # For OR expressions, file is usable if ANY license is permissive
            if ' OR ' in spdx_id:
                licenses = [l.strip() for l in spdx_id.split(' OR ')]
                return licenses
            # For AND expressions, ALL must be permissive
            if ' AND ' in spdx_id:
                licenses = [l.strip() for l in spdx_id.split(' AND ')]
                return licenses
            return [spdx_id]
    return None

def is_permissive_license(spdx_ids, expression_type='OR'):
    """
    Check if license identifiers are permissive.
    For OR: at least one must be permissive.
    For AND: all must be permissive.
    """
    if not spdx_ids:
        return None  # Unknown

    normalized = []
    for sid in spdx_ids:
        sid = SPDX_ALIASES.get(sid, sid)
        normalized.append(sid)

    if expression_type == 'OR':
        return any(sid in PERMISSIVE_SPDX for sid in normalized)
    else:  # AND
        return all(sid in PERMISSIVE_SPDX for sid in normalized)

2.3 Repository-Level License Detection

For repos without SPDX headers in individual files, we fall back to the repository-level LICENSE file:

import difflib

# Reference license texts (first 500 chars for matching)
LICENSE_TEMPLATES = {
    'MIT': 'permission is hereby granted, free of charge, to any person obtaining a copy',
    'Apache-2.0': 'apache license\nversion 2.0, january 2004',
    'BSD-2-Clause': 'redistribution and use in source and binary forms, with or without modification, are permitted',
    'BSD-3-Clause': 'redistribution and use in source and binary forms, with or without modification, are permitted provided that the following',
    'GPL-2.0': 'gnu general public license\nversion 2, june 1991',
    'GPL-3.0': 'gnu general public license\nversion 3, 29 june 2007',
    'AGPL-3.0': 'gnu affero general public license\nversion 3, 19 november 2007',
    'LGPL-2.1': 'gnu lesser general public license\nversion 2.1, february 1999',
    'Unlicense': 'this is free and unencumbered software released into the public domain',
    'ISC': 'permission to use, copy, modify, and/or distribute this software',
}

def detect_license_from_file(license_text):
    """
    Match license file content against known templates.
    Returns (spdx_id, confidence) or (None, 0.0).
    """
    normalized = license_text.lower().strip()

    best_match = None
    best_score = 0.0

    for spdx_id, template in LICENSE_TEMPLATES.items():
        # Check if template appears in the license text
        if template in normalized:
            # Direct substring match: high confidence
            return spdx_id, 0.95

        # Fuzzy matching on first 500 chars
        ratio = difflib.SequenceMatcher(
            None, normalized[:500], template
        ).ratio()
        if ratio > best_score:
            best_score = ratio
            best_match = spdx_id

    if best_score > 0.7:
        return best_match, best_score
    return None, 0.0

def filter_repository(repo_files, license_text=None):
    """
    Filter a repository: keep only if license is permissive.
    Returns (keep, license_id, confidence).
    """
    # 1. Try repo-level license file
    if license_text:
        license_id, confidence = detect_license_from_file(license_text)
        if license_id and confidence > 0.7:
            is_permissive = license_id in PERMISSIVE_SPDX or any(
                alias == license_id
                for alias, normalized in SPDX_ALIASES.items()
                if normalized in PERMISSIVE_SPDX
            )
            return is_permissive, license_id, confidence

    # 2. Try SPDX headers in files (majority vote)
    spdx_votes = {}
    for file_content in repo_files[:100]:  # Sample first 100 files
        spdx_ids = extract_spdx_from_header(file_content)
        if spdx_ids:
            for sid in spdx_ids:
                spdx_votes[sid] = spdx_votes.get(sid, 0) + 1

    if spdx_votes:
        most_common = max(spdx_votes, key=spdx_votes.get)
        normalized = SPDX_ALIASES.get(most_common, most_common)
        return normalized in PERMISSIVE_SPDX, normalized, 0.8

    # 3. No license detected: conservative approach = exclude
    return False, None, 0.0

2.4 Drop Rate

License filtering is the single largest reduction step in code curation:

---

3. Deduplication

Code on GitHub is massively duplicated. Forks, vendored dependencies, copy-pasted snippets, boilerplate, and auto-generated code create a dataset where the same content appears hundreds or thousands of times. Training on duplicates wastes compute, memorizes specific sequences (increasing verbatim reproduction risk), and skews the distribution toward popular but not necessarily high-quality code.

3.1 Exact Deduplication (Hash-Based)

The simplest form: hash each file’s content and keep only one copy per hash.

import hashlib
from collections import defaultdict

def exact_dedup_files(file_iterator):
    """
    Exact file-level deduplication using SHA-256.
    Yields unique files.
    """
    seen_hashes = set()
    stats = {'total': 0, 'unique': 0, 'duplicate': 0}

    for file_path, content in file_iterator:
        stats['total'] += 1

        # Normalize: strip trailing whitespace, normalize line endings
        normalized = content.strip().replace('\r\n', '\n')
        content_hash = hashlib.sha256(normalized.encode('utf-8')).hexdigest()

        if content_hash not in seen_hashes:
            seen_hashes.add(content_hash)
            stats['unique'] += 1
            yield file_path, content
        else:
            stats['duplicate'] += 1

    print(f"Exact dedup: {stats['total']} -> {stats['unique']} "
          f"({stats['duplicate']} duplicates removed, "
          f"{stats['duplicate']/stats['total']:.1%} reduction)")

Exact deduplication typically removes 10-20% of files. It catches identical copies but misses near-duplicates (files that differ by whitespace, comments, variable names, or minor edits).

3.2 Near-Deduplication with MinHash

Near-deduplication finds files that are substantially similar but not identical. The standard approach is MinHash with Locality-Sensitive Hashing (LSH).

Step 1: Shingling — Convert each document to a set of n-grams (shingles).

Step 2: MinHash — Compute a fixed-size signature for each document’s shingle set. Two documents with high Jaccard similarity will have similar MinHash signatures.

Step 3: LSH Banding — Group documents with similar signatures into buckets. Documents in the same bucket are candidate duplicates.

import struct
import random

class MinHasher:
    """
    MinHash signature generator for near-duplicate detection.
    """

    def __init__(self, num_hashes=128, ngram_size=5):
        self.num_hashes = num_hashes
        self.ngram_size = ngram_size
        # Generate random hash functions: h(x) = (a*x + b) mod p
        self.max_hash = (1 << 32) - 1
        self.prime = 4294967311  # Large prime > 2^32
        random.seed(42)
        self.hash_params = [
            (random.randint(1, self.prime - 1), random.randint(0, self.prime - 1))
            for _ in range(num_hashes)
        ]

    def get_shingles(self, text):
        """Extract character n-grams from text."""
        # For code: use token-level shingles (split on whitespace/punctuation)
        tokens = re.findall(r'\w+|[^\w\s]', text)
        shingles = set()
        for i in range(len(tokens) - self.ngram_size + 1):
            shingle = ' '.join(tokens[i:i + self.ngram_size])
            shingles.add(shingle)
        return shingles

    def compute_signature(self, text):
        """Compute MinHash signature for a document."""
        shingles = self.get_shingles(text)
        if not shingles:
            return [self.max_hash] * self.num_hashes

        # Hash each shingle
        shingle_hashes = []
        for s in shingles:
            h = struct.unpack('<I', hashlib.md5(s.encode()).digest()[:4])[0]
            shingle_hashes.append(h)

        # For each hash function, find minimum
        signature = []
        for a, b in self.hash_params:
            min_val = self.max_hash
            for h in shingle_hashes:
                val = (a * h + b) % self.prime
                if val < min_val:
                    min_val = val
            signature.append(min_val)

        return signature

    def estimate_similarity(self, sig1, sig2):
        """Estimate Jaccard similarity from MinHash signatures."""
        matches = sum(1 for a, b in zip(sig1, sig2) if a == b)
        return matches / len(sig1)


class LSHIndex:
    """
    Locality-Sensitive Hashing for finding near-duplicate candidates.
    """

    def __init__(self, num_hashes=128, num_bands=16):
        """
        num_bands: Number of LSH bands. Each band has num_hashes/num_bands rows.
        More bands = higher recall, more false positives.
        """
        self.num_bands = num_bands
        self.rows_per_band = num_hashes // num_bands
        # One hash table per band
        self.buckets = [defaultdict(list) for _ in range(num_bands)]

    def insert(self, doc_id, signature):
        """Insert a document's MinHash signature into the LSH index."""
        for band_idx in range(self.num_bands):
            start = band_idx * self.rows_per_band
            end = start + self.rows_per_band
            band_hash = hashlib.md5(
                str(signature[start:end]).encode()
            ).hexdigest()
            self.buckets[band_idx][band_hash].append(doc_id)

    def query(self, signature):
        """Find candidate near-duplicates for a signature."""
        candidates = set()
        for band_idx in range(self.num_bands):
            start = band_idx * self.rows_per_band
            end = start + self.rows_per_band
            band_hash = hashlib.md5(
                str(signature[start:end]).encode()
            ).hexdigest()
            for doc_id in self.buckets[band_idx][band_hash]:
                candidates.add(doc_id)
        return candidates

3.3 Function-Level Deduplication

File-level deduplication misses a critical pattern: the same function copied across many files. A utility function for parsing JSON, a common algorithm implementation, or a standard error handler might appear in thousands of files that are otherwise unique.

Function-level deduplication extracts individual functions, deduplicates at the function level, and then flags files where more than a threshold of content (e.g., 70%) consists of duplicated functions.

import ast

def extract_python_functions(source_code):
    """
    Extract function bodies from Python source code.
    Returns list of (function_name, function_source).
    """
    try:
        tree = ast.parse(source_code)
    except SyntaxError:
        return []

    functions = []
    lines = source_code.split('\n')

    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            start_line = node.lineno - 1
            end_line = node.end_lineno
            func_source = '\n'.join(lines[start_line:end_line])
            functions.append((node.name, func_source))

    return functions

def function_level_dedup(files, minhash_threshold=0.8):
    """
    Deduplicate at function level using MinHash.
    Flag files where most content is duplicated functions.
    """
    minhasher = MinHasher(num_hashes=64, ngram_size=3)
    lsh = LSHIndex(num_hashes=64, num_bands=8)
    func_signatures = {}  # func_id -> signature
    func_id_counter = 0

    # Phase 1: Extract and index all functions
    file_functions = {}  # file_path -> [(func_name, func_id, is_duplicate)]

    for file_path, content in files:
        functions = extract_python_functions(content)
        file_func_info = []

        for func_name, func_source in functions:
            sig = minhasher.compute_signature(func_source)
            candidates = lsh.query(sig)

            is_duplicate = False
            for candidate_id in candidates:
                if candidate_id in func_signatures:
                    similarity = minhasher.estimate_similarity(
                        sig, func_signatures[candidate_id]
                    )
                    if similarity >= minhash_threshold:
                        is_duplicate = True
                        break

            func_id = func_id_counter
            func_id_counter += 1
            func_signatures[func_id] = sig
            lsh.insert(func_id, sig)
            file_func_info.append((func_name, func_id, is_duplicate))

        file_functions[file_path] = file_func_info

    # Phase 2: Score files by duplicate function ratio
    results = {}
    for file_path, funcs in file_functions.items():
        if not funcs:
            results[file_path] = 0.0
            continue
        dup_count = sum(1 for _, _, is_dup in funcs if is_dup)
        results[file_path] = dup_count / len(funcs)

    return results

3.4 Combined Deduplication Pipeline

def dedup_pipeline(file_iterator, exact_threshold=True, near_threshold=0.85):
    """
    Full deduplication pipeline: exact -> near -> function-level.
    """
    # Stage 1: Exact dedup (cheapest, run first)
    exact_unique = list(exact_dedup_files(file_iterator))
    print(f"After exact dedup: {len(exact_unique)} files")

    # Stage 2: Near-dedup with MinHash + LSH
    minhasher = MinHasher(num_hashes=128, ngram_size=5)
    lsh = LSHIndex(num_hashes=128, num_bands=16)
    signatures = {}
    near_unique = []
    cluster_map = {}  # doc_id -> cluster_id

    for idx, (file_path, content) in enumerate(exact_unique):
        sig = minhasher.compute_signature(content)
        candidates = lsh.query(sig)

        is_near_dup = False
        for candidate_idx in candidates:
            if candidate_idx != idx and candidate_idx in signatures:
                similarity = minhasher.estimate_similarity(sig, signatures[candidate_idx])
                if similarity >= near_threshold:
                    is_near_dup = True
                    break

        if not is_near_dup:
            near_unique.append((file_path, content))

        signatures[idx] = sig
        lsh.insert(idx, sig)

    print(f"After near dedup: {len(near_unique)} files")
    return near_unique

---

4. Quality Scoring

After deduplication, the remaining files vary enormously in quality. A well-tested library with documentation scores differently from a student’s abandoned homework assignment. Quality scoring assigns a numeric score to each file based on multiple signals, then filters based on a threshold.

4.1 Syntax Checking (Lint Score)

The most basic quality signal: does the code parse?

import ast
import subprocess
import json

def syntax_check_python(source_code):
    """
    Check if Python code parses without syntax errors.
    Returns (passes, error_message).
    """
    try:
        ast.parse(source_code)
        return True, None
    except SyntaxError as e:
        return False, str(e)

def lint_score_python(source_code):
    """
    Run pylint/ruff on source code and return a normalized score.
    Score in [0, 1] where 1 is perfect.
    """
    # First: must parse
    passes, error = syntax_check_python(source_code)
    if not passes:
        return 0.0

    # Write to temp file and run ruff
    import tempfile
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(source_code)
        temp_path = f.name

    try:
        result = subprocess.run(
            ['ruff', 'check', '--output-format=json', '--select=E,W,F', temp_path],
            capture_output=True, text=True, timeout=10,
        )
        if result.stdout:
            issues = json.loads(result.stdout)
            num_issues = len(issues)
        else:
            num_issues = 0

        # Normalize: 0 issues = 1.0, 10+ issues per 100 lines = 0.0
        num_lines = max(1, source_code.count('\n'))
        issues_per_100_lines = (num_issues / num_lines) * 100
        score = max(0.0, 1.0 - issues_per_100_lines / 10.0)
        return score
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return 0.5  # Neutral score if linter unavailable
    finally:
        Path(temp_path).unlink(missing_ok=True)

def syntax_check_generic(source_code, language):
    """
    Language-agnostic syntax check using tree-sitter.
    """
    # tree-sitter can parse 100+ languages without executing them
    try:
        import tree_sitter
        parser = get_parser_for_language(language)  # Assumed helper
        tree = parser.parse(bytes(source_code, 'utf8'))
        # Count ERROR nodes in the parse tree
        error_count = count_error_nodes(tree.root_node)
        return error_count == 0, error_count
    except Exception:
        return True, 0  # Assume valid if parser unavailable

4.2 Cyclomatic Complexity

Cyclomatic complexity measures the number of independent execution paths through a function. High complexity correlates with difficult-to-understand code, but very low complexity (no branches at all) might indicate trivial or boilerplate code.

$$V(G) = E - N + 2P$$

where $E$ is the number of edges in the control flow graph, $N$ is the number of nodes, and $P$ is the number of connected components (usually 1 for a single function).

In practice, count the decision points:

def cyclomatic_complexity_python(source_code):
    """
    Compute cyclomatic complexity for each function in Python source.
    Returns list of (function_name, complexity).
    """
    try:
        tree = ast.parse(source_code)
    except SyntaxError:
        return []

    results = []

    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            complexity = 1  # Base complexity
            for child in ast.walk(node):
                # Each decision point adds 1
                if isinstance(child, (ast.If, ast.While, ast.For, ast.AsyncFor)):
                    complexity += 1
                elif isinstance(child, ast.ExceptHandler):
                    complexity += 1
                elif isinstance(child, ast.BoolOp):
                    # 'and' and 'or' operators add (num_values - 1)
                    complexity += len(child.values) - 1
                elif isinstance(child, ast.Assert):
                    complexity += 1
                elif isinstance(child, (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)):
                    # Each comprehension adds at least 1
                    complexity += len(child.generators)

            results.append((node.name, complexity))

    return results

def file_complexity_score(source_code):
    """
    Score a file based on cyclomatic complexity distribution.
    Ideal: moderate complexity (3-15 per function).
    """
    complexities = cyclomatic_complexity_python(source_code)
    if not complexities:
        return 0.5  # No functions = neutral

    avg_complexity = sum(c for _, c in complexities) / len(complexities)
    max_complexity = max(c for _, c in complexities)

    # Score: penalize both trivial (avg below 2) and overly complex (avg above 20)
    if avg_complexity < 2:
        score = 0.3  # Likely trivial/boilerplate
    elif avg_complexity <= 15:
        score = 1.0  # Good range
    elif avg_complexity <= 30:
        score = 0.6  # Getting complex but acceptable
    else:
        score = 0.2  # Overly complex, likely poor quality

    # Heavy penalty for any function above 50
    if max_complexity > 50:
        score *= 0.5

    return score

4.3 Documentation Ratio

Well-documented code is typically higher quality. The documentation ratio measures the proportion of lines that are comments or docstrings:

def documentation_ratio(source_code, language='Python'):
    """
    Compute the ratio of documentation lines to total lines.
    """
    lines = source_code.split('\n')
    total_lines = len(lines)
    if total_lines == 0:
        return 0.0

    doc_lines = 0
    in_docstring = False

    for line in lines:
        stripped = line.strip()

        if language == 'Python':
            # Docstrings
            if '"""' in stripped or "'''" in stripped:
                doc_lines += 1
                count = stripped.count('"""') + stripped.count("'''")
                if count == 1:
                    in_docstring = not in_docstring
                continue
            if in_docstring:
                doc_lines += 1
                continue
            # Comments
            if stripped.startswith('#'):
                doc_lines += 1

        elif language in ('JavaScript', 'TypeScript', 'Java', 'C', 'C++', 'Go', 'Rust'):
            if stripped.startswith('//'):
                doc_lines += 1
            elif stripped.startswith('/*') or stripped.startswith('*'):
                doc_lines += 1
            elif stripped.endswith('*/'):
                doc_lines += 1

    ratio = doc_lines / total_lines

    # Score: 5-25% documentation is ideal
    if ratio < 0.02:
        return 0.2   # Almost no documentation
    elif ratio < 0.05:
        return 0.5
    elif ratio <= 0.30:
        return 1.0   # Good documentation
    elif ratio <= 0.50:
        return 0.7   # Might be overly commented
    else:
        return 0.3   # More comments than code (likely autogenerated headers)

4.4 Combined Quality Score

def compute_quality_score(source_code, language='Python'):
    """
    Combined quality score from multiple signals.
    Returns score in [0, 1].
    """
    weights = {
        'syntax': 0.30,
        'complexity': 0.25,
        'documentation': 0.20,
        'line_length': 0.15,
        'file_size': 0.10,
    }

    scores = {}

    # Syntax check
    passes, _ = syntax_check_python(source_code) if language == 'Python' else (True, None)
    scores['syntax'] = 1.0 if passes else 0.0

    # Complexity
    scores['complexity'] = file_complexity_score(source_code) if language == 'Python' else 0.5

    # Documentation
    scores['documentation'] = documentation_ratio(source_code, language)

    # Line length: penalize extremely long lines (generated code)
    lines = source_code.split('\n')
    long_lines = sum(1 for l in lines if len(l) > 200)
    long_ratio = long_lines / max(1, len(lines))
    scores['line_length'] = max(0.0, 1.0 - long_ratio * 5)

    # File size: very small files (under 10 lines) or very large (over 5000 lines) penalized
    num_lines = len(lines)
    if num_lines < 10:
        scores['file_size'] = 0.3
    elif num_lines <= 2000:
        scores['file_size'] = 1.0
    elif num_lines <= 5000:
        scores['file_size'] = 0.6
    else:
        scores['file_size'] = 0.2

    # Weighted combination
    total = sum(weights[k] * scores[k] for k in weights)
    return total, scores

# Example usage
code = '''
def merge_sort(arr):
    """
    Sort an array using merge sort algorithm.
    Time complexity: O(n log n)
    Space complexity: O(n)
    """
    if len(arr) <= 1:
        return arr

    mid = len(arr) // 2
    left = merge_sort(arr[:mid])
    right = merge_sort(arr[mid:])

    return merge(left, right)

def merge(left, right):
    """Merge two sorted arrays."""
    result = []
    i = j = 0
    while i < len(left) and j < len(right):
        if left[i] <= right[j]:
            result.append(left[i])
            i += 1
        else:
            result.append(right[j])
            j += 1
    result.extend(left[i:])
    result.extend(right[j:])
    return result
'''
score, breakdown = compute_quality_score(code, 'Python')
# score: ~0.85, breakdown: {syntax: 1.0, complexity: 1.0, documentation: 1.0, ...}

A common threshold is 0.5, which retains roughly 70% of files but removes the worst offenders.

---

5. Language Distribution

5.1 The Natural Distribution Problem

The raw distribution of code on GitHub is heavily skewed:

Training on the natural distribution means the model sees 100x more JavaScript than Haskell. The model will be excellent at JavaScript and mediocre at Haskell. Depending on target use cases, this may or may not be acceptable.

5.2 Resampling Strategies

Natural distribution: Use the raw proportions. Maximizes performance on popular languages.

Uniform distribution: Equal tokens per language. Maximizes breadth across all languages, but wastes capacity on obscure languages while undertraining on common ones.

Square-root sampling (StarCoder approach): Sample proportional to the square root of the natural proportion. This smooths the distribution without equalizing it:

$$p_\text{sampled}(L) = \frac{\sqrt{p_\text{natural}(L)}}{\sum_{L'} \sqrt{p_\text{natural}(L')}}$$

import math

def compute_sampling_weights(language_counts, strategy='sqrt'):
    """
    Compute sampling weights for each language.

    language_counts: dict of language -> file_count
    strategy: 'natural', 'uniform', 'sqrt', 'log'
    """
    total = sum(language_counts.values())
    natural_probs = {
        lang: count / total for lang, count in language_counts.items()
    }

    if strategy == 'natural':
        return natural_probs

    elif strategy == 'uniform':
        n = len(language_counts)
        return {lang: 1.0 / n for lang in language_counts}

    elif strategy == 'sqrt':
        raw = {lang: math.sqrt(p) for lang, p in natural_probs.items()}
        total_raw = sum(raw.values())
        return {lang: v / total_raw for lang, v in raw.items()}

    elif strategy == 'log':
        raw = {lang: math.log(1 + p * 1000) for lang, p in natural_probs.items()}
        total_raw = sum(raw.values())
        return {lang: v / total_raw for lang, v in raw.items()}

# Example with realistic counts
counts = {
    'JavaScript': 220_000_000,
    'Python': 160_000_000,
    'Java': 120_000_000,
    'TypeScript': 70_000_000,
    'C++': 60_000_000,
    'Go': 30_000_000,
    'Rust': 15_000_000,
    'Haskell': 2_000_000,
}

natural = compute_sampling_weights(counts, 'natural')
sqrt_weights = compute_sampling_weights(counts, 'sqrt')

# JavaScript: natural=32.5%, sqrt=22.8% (down-weighted)
# Haskell:   natural=0.3%,  sqrt=2.2%  (up-weighted 7x)

5.3 Chinchilla Scaling for Code

The Chinchilla scaling law ($N \approx 20D$ for optimal compute allocation, where $N$ is parameters and $D$ is tokens) applies to code with a key modification: code tokens are denser in information than natural language tokens. Empirically, 1 code token carries roughly 1.5x the information of 1 English token (lower perplexity per token).

This means a 7B parameter code model should be trained on approximately:

$$D_\text{code} = \frac{20 \times 7 \times 10^9}{1.5} \approx 93 \text{ billion tokens}$$

StarCoder 2 15B was trained on 600B+ tokens (significantly more than Chinchilla-optimal), reflecting the common practice of overtraining smaller models to improve inference efficiency.

---

6. PII Removal

Code on GitHub contains an alarming density of personally identifiable information and secrets: API keys, database passwords, email addresses, AWS credentials, private SSH keys, and authentication tokens.

6.1 Regex-Based Detection

import re

# PII patterns
PII_PATTERNS = {
    'email': re.compile(
        r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
    ),
    'ip_address': re.compile(
        r'\b(?:\d{1,3}\.){3}\d{1,3}\b',
    ),
    'aws_access_key': re.compile(
        r'(?:AKIA|ASIA)[A-Z0-9]{16}',
    ),
    'aws_secret_key': re.compile(
        r'(?:aws_secret_access_key|secret_key)\s*[=:]\s*["\']?[A-Za-z0-9/+=]{40}["\']?',
        re.IGNORECASE,
    ),
    'github_token': re.compile(
        r'gh[ps]_[A-Za-z0-9_]{36,}',
    ),
    'generic_api_key': re.compile(
        r'(?:api[_-]?key|apikey|api[_-]?secret|api[_-]?token)\s*[=:]\s*["\']?[A-Za-z0-9_\-]{20,}["\']?',
        re.IGNORECASE,
    ),
    'private_key_header': re.compile(
        r'-----BEGIN (?:RSA |DSA |EC |OPENSSH )?PRIVATE KEY-----',
    ),
    'jwt_token': re.compile(
        r'eyJ[A-Za-z0-9_-]{10,}\.eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}',
    ),
    'slack_token': re.compile(
        r'xox[bporas]-[A-Za-z0-9-]{10,}',
    ),
    'phone_us': re.compile(
        r'\b(?:\+1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b',
    ),
}

# Replacement tokens
REPLACEMENTS = {
    'email': 'EMAIL_REDACTED@example.com',
    'ip_address': '0.0.0.0',
    'aws_access_key': 'AKIAIOSFODNN7EXAMPLE',
    'aws_secret_key': 'api_key = "REDACTED_SECRET_KEY"',
    'github_token': 'ghp_REDACTED_TOKEN_PLACEHOLDER',
    'generic_api_key': 'api_key = "REDACTED_API_KEY"',
    'private_key_header': '-----BEGIN REDACTED PRIVATE KEY-----',
    'jwt_token': 'eyJREDACTED.eyJREDACTED.REDACTED',
    'slack_token': 'xoxb-REDACTED-TOKEN',
    'phone_us': '555-000-0000',
}

def redact_pii(source_code):
    """
    Scan source code for PII and secrets, replace with safe placeholders.
    Returns (redacted_code, detections).
    """
    redacted = source_code
    detections = []

    for pii_type, pattern in PII_PATTERNS.items():
        matches = list(pattern.finditer(redacted))
        if matches:
            replacement = REPLACEMENTS.get(pii_type, 'REDACTED')
            for match in reversed(matches):  # Reverse to preserve indices
                detections.append({
                    'type': pii_type,
                    'start': match.start(),
                    'end': match.end(),
                    'value_length': match.end() - match.start(),
                })
                redacted = (
                    redacted[:match.start()]
                    + replacement
                    + redacted[match.end():]
                )

    return redacted, detections

6.2 False Positive Management

The biggest challenge with regex-based PII detection is false positives. The email regex matches user@localhost in test fixtures. The IP address regex matches version strings like 1.2.3.4. The API key regex matches base64-encoded test data.

def filter_false_positives(detections, source_code):
    """
    Reduce false positives by checking context.
    """
    filtered = []

    for detection in detections:
        pii_type = detection['type']

        # Check if in a test file
        if pii_type == 'email' and 'test' in source_code[:200].lower():
            # Test files often have fake emails - still redact but lower confidence
            detection['confidence'] = 0.5
        elif pii_type == 'ip_address':
            # Skip common non-PII IPs
            start, end = detection['start'], detection['end']
            ip = source_code[start:end]
            safe_ips = {'127.0.0.1', '0.0.0.0', '255.255.255.0', '255.255.255.255'}
            if ip in safe_ips:
                continue
            # Check if it is a version string (preceded by 'v' or 'version')
            context_before = source_code[max(0, start-20):start].lower()
            if 'version' in context_before or context_before.rstrip().endswith('v'):
                continue
            detection['confidence'] = 0.7
        else:
            detection['confidence'] = 0.9

        filtered.append(detection)

    return filtered

6.3 ML-Based Secret Detection

For higher accuracy, train a classifier on labeled examples of real secrets vs false positives. GitHub’s own secret scanning uses this approach:

# Simplified feature extraction for secret classifier
def extract_secret_features(candidate_string, context):
    """
    Extract features for ML-based secret classification.
    """
    features = {
        # String entropy (secrets have high entropy)
        'entropy': compute_shannon_entropy(candidate_string),
        # Length
        'length': len(candidate_string),
        # Character class distribution
        'uppercase_ratio': sum(c.isupper() for c in candidate_string) / max(1, len(candidate_string)),
        'digit_ratio': sum(c.isdigit() for c in candidate_string) / max(1, len(candidate_string)),
        'special_ratio': sum(not c.isalnum() for c in candidate_string) / max(1, len(candidate_string)),
        # Context features
        'near_assignment': '=' in context or ':' in context,
        'in_string_literal': context.count('"') % 2 == 1 or context.count("'") % 2 == 1,
        'near_secret_keyword': any(
            kw in context.lower()
            for kw in ['secret', 'key', 'token', 'password', 'api_key', 'credential']
        ),
        'in_comment': context.lstrip().startswith('#') or context.lstrip().startswith('//'),
        'in_test_file': 'test' in context[:100].lower(),
    }
    return features

def compute_shannon_entropy(s):
    """Compute Shannon entropy of a string."""
    if not s:
        return 0.0
    freq = {}
    for c in s:
        freq[c] = freq.get(c, 0) + 1
    length = len(s)
    entropy = 0.0
    for count in freq.values():
        p = count / length
        if p > 0:
            entropy -= p * math.log2(p)
    return entropy

---

7. The Complete Pipeline

Putting all stages together:

def code_curation_pipeline(raw_files):
    """
    Complete code dataset curation pipeline.
    Input: iterator of (file_path, content, repo_license) tuples.
    Output: curated dataset.
    """
    stats = {
        'input': 0,
        'after_basic_filter': 0,
        'after_license': 0,
        'after_exact_dedup': 0,
        'after_near_dedup': 0,
        'after_quality': 0,
        'after_pii': 0,
    }

    # Stage 1: Basic filtering (file size, extensions, paths)
    def basic_filter(files):
        for path, content, license_info in files:
            stats['input'] += 1
            if should_include_file(path, len(content.encode('utf-8'))):
                stats['after_basic_filter'] += 1
                yield path, content, license_info

    # Stage 2: License filtering
    def license_filter(files):
        for path, content, license_info in files:
            is_permissive, _, _ = filter_repository([content], license_info)
            if is_permissive:
                stats['after_license'] += 1
                yield path, content

    # Stage 3: Exact deduplication
    def exact_dedup(files):
        seen = set()
        for path, content in files:
            normalized = content.strip().replace('\r\n', '\n')
            h = hashlib.sha256(normalized.encode()).hexdigest()
            if h not in seen:
                seen.add(h)
                stats['after_exact_dedup'] += 1
                yield path, content

    # Stage 4: Quality scoring
    def quality_filter(files, threshold=0.5):
        for path, content in files:
            lang = detect_language(path, content)
            score, _ = compute_quality_score(content, lang)
            if score >= threshold:
                stats['after_quality'] += 1
                yield path, content

    # Stage 5: PII removal
    def pii_stage(files):
        for path, content in files:
            redacted, detections = redact_pii(content)
            stats['after_pii'] += 1
            yield path, redacted

    # Chain the stages
    pipeline = pii_stage(
        quality_filter(
            exact_dedup(
                license_filter(
                    basic_filter(raw_files)
                )
            )
        )
    )

    results = list(pipeline)

    print("=== Pipeline Statistics ===")
    for stage, count in stats.items():
        print(f"  {stage}: {count:,}")

    return results

The final dataset is 18% of the original input by file count. By token count, the reduction is even larger (roughly 12% of original tokens) because the removed files tend to be large generated files, duplicated libraries, and low-quality boilerplate.

---

8. Scaling Considerations

8.1 Distributed Deduplication

MinHash + LSH on 3 billion files does not fit in memory on a single machine. The standard approach is MapReduce-style distributed processing:

# Distributed MinHash dedup (pseudocode for Spark / Ray)
def distributed_near_dedup(file_rdd, num_hashes=128, num_bands=16):
    """
    Distributed near-dedup using MinHash + LSH on a cluster.
    """
    rows_per_band = num_hashes // num_bands

    # Step 1: Compute MinHash signatures (embarrassingly parallel)
    signatures_rdd = file_rdd.map(
        lambda file: (file.id, minhasher.compute_signature(file.content))
    )

    # Step 2: For each band, emit (band_hash, doc_id) pairs
    def emit_band_hashes(doc_id, signature):
        for band_idx in range(num_bands):
            start = band_idx * rows_per_band
            end = start + rows_per_band
            band_key = (band_idx, hash(tuple(signature[start:end])))
            yield band_key, doc_id

    band_pairs_rdd = signatures_rdd.flatMap(
        lambda x: emit_band_hashes(x[0], x[1])
    )

    # Step 3: Group by band hash to find candidate clusters
    clusters_rdd = band_pairs_rdd.groupByKey().filter(
        lambda x: len(list(x[1])) > 1  # Only groups with duplicates
    )

    # Step 4: Union-Find to merge clusters across bands
    # (A doc might appear in multiple band buckets)
    duplicate_pairs = clusters_rdd.flatMap(
        lambda x: [(sorted(list(x[1]))[0], doc_id) for doc_id in list(x[1])[1:]]
    )

    # Step 5: Keep only cluster representatives
    duplicates_to_remove = set(duplicate_pairs.map(lambda x: x[1]).collect())

    return file_rdd.filter(lambda f: f.id not in duplicates_to_remove)

8.2 Processing Time Estimates

Total pipeline time: approximately 155 hours (6.5 days) on a 128-core cluster. The dominant costs are near-deduplication (shuffle-intensive) and quality scoring (requires parsing and analysis). This is why the pipeline order matters: cheap filters (basic, license, exact dedup) run first to reduce the dataset before the expensive stages.