Overview

This guide enables you to access Historical Price Data files from the cfg-public-proper-wallaby S3 bucket (eu-west-1) configured with Requester Pays policy, ensuring optimal download performance and system stability.

Prerequisites

• AWS CLI installed and configured
• Valid AWS credentials with appropriate permissions
• Access to eu-west-1 region
• Understanding of requester pays billing model

Configuration Steps

1. Set AWS Credentials & Region

export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_DEFAULT_REGION="eu-west-1"

2. Verify Bucket Access

aws s3 ls s3://cfg-public-proper-wallaby/ \
  --region eu-west-1 \
  --request-payer requester

3. Discover Available Instruments

Before downloading, list all currency pairs (instruments) available in the bucket. Use --delimiter / (or plain aws s3 ls) to list only top-level prefixes — avoid recursive listing, which would scan every file just to find the ~20-30 folder names.

Quick list (simplest):

aws s3 ls s3://cfg-public-proper-wallaby/ \
  --region eu-west-1 \
  --request-payer requester

Clean list (names only, sorted, saved to file):

aws s3api list-objects-v2 \
  --bucket cfg-public-proper-wallaby \
  --delimiter "/" \
  --region eu-west-1 \
  --request-payer requester \
  --query "CommonPrefixes[].Prefix" \
  --output text | tr '\t' '\n' | sed 's|/$||' | sort > instruments.txt

echo "Total instruments: $(wc -l < instruments.txt)"
cat instruments.txt

Python (Boto3):

import boto3
from botocore.config import Config

config = Config(region_name='eu-west-1')
s3_client = boto3.client('s3', config=config)

def list_instruments(bucket):
    """Get all top-level instrument prefixes (currency pairs)"""
    instruments = []
    paginator = s3_client.get_paginator('list_objects_v2')

    page_iterator = paginator.paginate(
        Bucket=bucket,
        Delimiter='/',
        RequestPayer='requester'
    )

    for page in page_iterator:
        for prefix in page.get('CommonPrefixes', []):
            instrument = prefix['Prefix'].rstrip('/')
            instruments.append(instrument)

    return sorted(instruments)

if __name__ == "__main__":
    BUCKET = 'cfg-public-proper-wallaby'
    instruments = list_instruments(BUCKET)

    print(f"Total instruments found: {len(instruments)}\n")
    for inst in instruments:
        print(inst)

    with open('instruments.txt', 'w') as f:
        f.write('\n'.join(instruments))

Usage:

python3 list_instruments.py

⚠️ Performance & Cost Note: Using --delimiter / (or Delimiter='/' in Boto3) is essential — it triggers a lightweight CommonPrefixes response instead of enumerating every file. This costs only a few cheap LIST requests and completes in seconds, versus a slow, costlier full-bucket scan without the delimiter.

4. Basic S3 Access Command with Requester Pays

# Download single file
aws s3 cp s3://cfg-public-proper-wallaby/EURUSD/file.json . \
  --region eu-west-1 \
  --request-payer requester

5. Batch Download with Performance Optimization

# Download all history for specific currency pair
aws s3 sync s3://cfg-public-proper-wallaby/EURUSD/ ./EURUSD-data/ \
  --region eu-west-1 \
  --request-payer requester \
  --no-progress \
  --max-concurrent-requests 20 \
  --max-bandwidth 100MB/s

Performance Optimization Parameters

6. Advanced Download Script (Speed & Stability)

#!/bin/bash

BUCKET_NAME="cfg-public-proper-wallaby"
REGION="eu-west-1"
CURRENCY_PAIR="${1:-EURUSD}"  # Default to EURUSD, or pass as argument
DESTINATION="./data/${CURRENCY_PAIR}/"
MAX_RETRIES=3
CONCURRENT_JOBS=20

mkdir -p "$DESTINATION"

LOG_FILE="s3-sync-${CURRENCY_PAIR}-$(date +%Y%m%d_%H%M%S).log"

echo "Starting S3 sync for $CURRENCY_PAIR from $BUCKET_NAME in region $REGION..." | tee "$LOG_FILE"

for attempt in $(seq 1 $MAX_RETRIES); do
    echo "Download attempt $attempt of $MAX_RETRIES..." | tee -a "$LOG_FILE"

    aws s3 sync "s3://$BUCKET_NAME/$CURRENCY_PAIR/" "$DESTINATION" \
      --region "$REGION" \
      --request-payer requester \
      --max-concurrent-requests 20 \
      --only-show-errors \
      --delete >> "$LOG_FILE" 2>&1

    EXIT_CODE=$?

    if [ $EXIT_CODE -eq 0 ]; then
        echo "✓ Download completed successfully at $(date)" | tee -a "$LOG_FILE"
        echo "Total files: $(find $DESTINATION -type f | wc -l)" | tee -a "$LOG_FILE"
        echo "Total size: $(du -sh $DESTINATION | cut -f1)" | tee -a "$LOG_FILE"
        exit 0
    else
        echo "✗ Download failed with exit code $EXIT_CODE. Retrying in 30 seconds..." | tee -a "$LOG_FILE"
        sleep 30
    fi
done

echo "✗ Download failed after $MAX_RETRIES attempts" | tee -a "$LOG_FILE"
exit 1

Usage:

chmod +x download-price-history.sh

./download-price-history.sh EURUSD
./download-price-history.sh GBPUSD
./download-price-history.sh

7. Batch Download Multiple Currency Pairs

#!/bin/bash

BUCKET_NAME="cfg-public-proper-wallaby"
REGION="eu-west-1"
MAX_RETRIES=3

PAIRS=("EURUSD" "GBPUSD" "USDJPY" "AUDUSD" "NZDUSD" "USDCAD")

for PAIR in "${PAIRS[@]}"; do
    echo "Starting download for $PAIR..."

    DESTINATION="./data/${PAIR}/"
    mkdir -p "$DESTINATION"

    for attempt in $(seq 1 $MAX_RETRIES); do
        aws s3 sync "s3://$BUCKET_NAME/$PAIR/" "$DESTINATION" \
          --region "$REGION" \
          --request-payer requester \
          --max-concurrent-requests 20 \
          --only-show-errors \
          --delete

        if [ $? -eq 0 ]; then
            echo "✓ $PAIR completed"
            break
        else
            echo "⚠ $PAIR attempt $attempt failed, retrying..."
            sleep 30
        fi
    done
done

echo "✓ All downloads completed"

8. Using Python Boto3 (Programmatic Access)

import boto3
import concurrent.futures
from botocore.config import Config
import logging
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

config = Config(
    max_pool_connections=20,
    retries={'max_attempts': 3, 'mode': 'adaptive'},
    connect_timeout=5,
    read_timeout=60,
    region_name='eu-west-1'
)

s3_client = boto3.client('s3', config=config)
s3_resource = boto3.resource('s3', config=config)

def download_file(bucket, key, filename):
    """Download file with requester pays enabled"""
    try:
        Path(filename).parent.mkdir(parents=True, exist_ok=True)
        s3_client.download_file(
            Bucket=bucket,
            Key=key,
            Filename=filename,
            ExtraArgs={'RequestPayer': 'requester'}
        )
        logger.info(f"✓ Downloaded: {key}")
        return True
    except Exception as e:
        logger.error(f"✗ Error downloading {key}: {e}")
        return False

def batch_download(bucket, currency_pair, destination, max_workers=10):
    """Download all files for a currency pair in parallel"""
    try:
        bucket_obj = s3_resource.Bucket(bucket)

        # Trailing slash on the prefix avoids accidentally matching a
        # different instrument that happens to share this one as a prefix
        # (e.g. "EURUSD" vs. a hypothetical "EURUSDT").
        prefix = f"{currency_pair}/"
        objects = list(bucket_obj.objects.filter(Prefix=prefix, RequestPayer='requester'))
        logger.info(f"Found {len(objects)} objects to download for {currency_pair}")

        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = []

            for obj in objects:
                if obj.key.endswith('/'):  # Skip directories
                    continue

                filename = f"{destination}/{obj.key}"
                futures.append(executor.submit(download_file, bucket, obj.key, filename))

            completed = 0
            failed = 0
            for future in concurrent.futures.as_completed(futures):
                if future.result():
                    completed += 1
                else:
                    failed += 1

            logger.info(f"\n=== Download Summary ===")
            logger.info(f"Currency Pair: {currency_pair}")
            logger.info(f"Total files: {len(futures)}")
            logger.info(f"Successful: {completed}")
            logger.info(f"Failed: {failed}")

            return completed, failed

    except Exception as e:
        logger.error(f"Batch download error: {e}")
        return 0, len(objects)

# Usage - Single Currency Pair
if __name__ == "__main__":
    BUCKET = 'cfg-public-proper-wallaby'
    CURRENCY_PAIR = 'EURUSD'  # Change as needed
    DESTINATION = f'./data/{CURRENCY_PAIR}'

    logger.info(f"Starting download from s3://{BUCKET}/{CURRENCY_PAIR}/")
    logger.info(f"Region: eu-west-1")
    logger.info(f"Destination: {DESTINATION}")

    completed, failed = batch_download(
        bucket=BUCKET,
        currency_pair=CURRENCY_PAIR,
        destination=DESTINATION,
        max_workers=10
    )

    if failed == 0:
        logger.info("\n✓ All files downloaded successfully!")
    else:
        logger.warning(f"\n⚠ {failed} files failed to download")

Usage:

python3 download_price_history.py
# Edit CURRENCY_PAIR variable for different pairs

Note: Given EUR/USD's confirmed 26,586 files, the list(bucket_obj.objects.filter(...)) call above will enumerate all of them via paginated ListObjectsV2 calls before any downloads start. This listing step incurs its own (small) request cost, billed separately from GET requests. For very large prefixes, consider paginating manually or testing with partial fetches first.

9. Download All Currency Pairs (Python)

import boto3
import concurrent.futures
from botocore.config import Config
import logging
from pathlib import Path

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

config = Config(
    max_pool_connections=20,
    retries={'max_attempts': 3, 'mode': 'adaptive'},
    connect_timeout=5,
    read_timeout=60,
    region_name='eu-west-1'
)

s3_client = boto3.client('s3', config=config)
s3_resource = boto3.resource('s3', config=config)

def download_file(bucket, key, filename):
    """Download file with requester pays enabled"""
    try:
        Path(filename).parent.mkdir(parents=True, exist_ok=True)
        s3_client.download_file(
            Bucket=bucket,
            Key=key,
            Filename=filename,
            ExtraArgs={'RequestPayer': 'requester'}
        )
        return True
    except Exception as e:
        logger.error(f"✗ Error downloading {key}: {e}")
        return False

def download_all_pairs(bucket, destination_base, max_workers=10):
    """Download all history for all currency pairs"""
    try:
        bucket_obj = s3_resource.Bucket(bucket)

        pairs = set()
        for obj in bucket_obj.objects.filter(RequestPayer='requester'):
            pair = obj.key.split('/')[0]
            if pair and not obj.key.endswith('/'):
                pairs.add(pair)

        logger.info(f"Found {len(pairs)} currency pairs: {sorted(pairs)}")

        total_completed = 0
        total_failed = 0

        for pair in sorted(pairs):
            logger.info(f"\n=== Processing {pair} ===")

            # Trailing slash keeps this scoped to exactly this instrument's keys
            prefix = f"{pair}/"
            pair_objects = list(bucket_obj.objects.filter(Prefix=prefix, RequestPayer='requester'))
            logger.info(f"Found {len(pair_objects)} objects for {pair}")

            with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
                futures = []

                for obj in pair_objects:
                    if obj.key.endswith('/'):
                        continue

                    filename = f"{destination_base}/{obj.key}"
                    futures.append(executor.submit(download_file, bucket, obj.key, filename))

                completed = sum(1 for f in concurrent.futures.as_completed(futures) if f.result())
                failed = len(futures) - completed

                total_completed += completed
                total_failed += failed

                logger.info(f"✓ {pair}: {completed}/{len(futures)} files downloaded")

        logger.info(f"\n=== FINAL SUMMARY ===")
        logger.info(f"Total Completed: {total_completed}")
        logger.info(f"Total Failed: {total_failed}")

        return total_completed, total_failed

    except Exception as e:
        logger.error(f"Error: {e}")
        return 0, 0

# Usage
if __name__ == "__main__":
    BUCKET = 'cfg-public-proper-wallaby'
    DESTINATION_BASE = './data'

    logger.info(f"Starting full archive download from s3://{BUCKET}/")
    logger.info(f"Region: eu-west-1")
    logger.info(f"Destination: {DESTINATION_BASE}")

    download_all_pairs(
        bucket=BUCKET,
        destination_base=DESTINATION_BASE,
        max_workers=10
    )

10. Quick Reference Commands

# List all currency pairs available
aws s3 ls s3://cfg-public-proper-wallaby/ \
  --region eu-west-1 \
  --request-payer requester

# Download EURUSD history
aws s3 sync s3://cfg-public-proper-wallaby/EURUSD/ ./EURUSD/ \
  --region eu-west-1 \
  --request-payer requester \
  --max-concurrent-requests 25

# Download GBPUSD history
aws s3 sync s3://cfg-public-proper-wallaby/GBPUSD/ ./GBPUSD/ \
  --region eu-west-1 \
  --request-payer requester \
  --max-concurrent-requests 25

# Count files in EURUSD
aws s3 ls s3://cfg-public-proper-wallaby/EURUSD/ \
  --region eu-west-1 \
  --request-payer requester \
  --recursive | wc -l

# Get total size of EURUSD
aws s3 ls s3://cfg-public-proper-wallaby/EURUSD/ \
  --region eu-west-1 \
  --request-payer requester \
  --recursive --summarize | grep "Total Size"

Key Motivation Points

Speed Enhancement ⚡

• Parallel Processing: 20-25 concurrent connections maximize throughput
• EU-West-1 Optimization: No cross-region latency
• Adaptive Retry Strategy: Intelligent failure recovery
• Connection Pooling: Minimizes overhead between requests
• Bandwidth Control: Prevents network saturation and stability issues

Stability Improvement

• Retry Mechanism: 3 attempts with 30-second intervals
• Error Logging: Detailed logs for troubleshooting
• Connection Timeout: 5-second connect, 60-second read timeout
• Adaptive Mode: Adjusts retry strategy based on error types
• Integrity Checks: Validates file downloads
• Graceful Degradation: Continues on partial failures

Cost Estimation

Pricing Structure

AWS S3 GET Request Pricing (Requester Pays, eu-west-1):

• $0.0004 per 1,000 requests
• $0.02 per GB data transfer (eu-west-1 egress rate)

The decode and pipeline steps in Sections 11–12 run entirely on your local machine — they add no additional AWS charges beyond the S3 download costs covered here.

Full Archive Estimate

⚠️ Full Archive: ~400 GB, ~20,000,000 files

Request cost  = (20,000,000 / 1,000) × $0.0004 = $8.00
Transfer cost = 400 × $0.02                     = $8.00
─────────────────────────────────────────────────────────
TOTAL                                           = $16.00

Archive-wide average file size: 400 GB / 20,000,000 files ≈ 20.97 KB/file

Real-World Example: EUR/USD (Verified Actual Data)

Confirmed via aws s3 ls --summarize:

Total objects: 26,586
Total size:    2.6 GB
Average file size: 2.6 GB / 26,586 ≈ 100.1 KB/file

Request Costs:

Number of requests: 26,586
Cost per 1,000 requests: $0.0004

Total request cost = (26,586 / 1,000) × $0.0004
                   = 26.586 × $0.0004
                   ≈ $0.0106

Data Transfer Costs:

Data size: 2.6 GB
Cost per GB: $0.02
Total transfer cost = 2.6 × $0.02 = $0.052

Total EUR/USD Download Cost (Verified):

Request costs:        $0.0106
Data transfer costs:  $0.052
────────────────────────────
TOTAL:                ~$0.06

Cost Comparison Table

Estimates for "5 Pairs" and "10 Pairs" assume file count/size similar to EUR/USD. Actual costs will vary by pair — majors may have more history/files than EUR/USD, exotics likely fewer.

Cost Optimization Tips

• ✓ Download only required currency pairs (not entire archive)
• ✓ Always verify actual size/file count per pair with --summarize — EUR/USD's real average file size (~100 KB) is ~5x the archive-wide average (~21 KB), showing averages can mislead
• ✓ Batch downloads to minimize request overhead
• ✓ Use --delete flag to avoid re-downloading existing files
• ✓ At ~$0.06 per pair (EUR/USD-verified), downloading dozens of pairs individually remains a small fraction of the $16 full-archive cost
• ✓ Use Section 3 (Discover Available Instruments) to confirm exact pair names before running batch scripts

Recommended Action Before Bulk Downloads

aws s3 ls s3://cfg-public-proper-wallaby/<PAIR>/ \
  --region eu-west-1 \
  --request-payer requester \
  --recursive --summarize | tail -3

This returns the exact Total Objects and Total Size, letting you compute an accurate cost via:

Request cost  = (files / 1000) × $0.0004
Transfer cost = size_in_GB × $0.02

Decoding .bi5 Files into Readable Tick Data

Dukascopy's .bi5 files are LZMA-compressed binary tick files. With the current bucket structure, each file represents one full day of tick data per instrument.

File Structure

Compression: Raw LZMA stream (not .xz container format)

Path convention (daily):

SYMBOL/YEAR/MONTH/DAY_ticks.bi5

Example: EURUSD/2024/00/15_ticks.bi5 → EUR/USD ticks for Jan 15, 2024

⚠️ Month is zero-indexed (January = 00, December = 11).

✅ No empty files: A missing file for a given day means no ticks were recorded (weekends, holidays). Handle missing S3 keys / FileNotFoundError as "no data" — not an error.

Decompressed Record Format

Each tick is a fixed 20-byte binary record, big-endian:

Price Scaling ("Point Value")

⚠️ The helper functions below only distinguish "JPY pair" vs. "everything else (100,000)." For indices, commodities, or any non-FX instrument, confirm the correct point value with the data provider before decoding — silently applying 100,000 to a non-FX instrument will produce wrong prices with no error raised. The updated get_point_value() in this guide now logs a warning when it falls back to the default for an unrecognized, non-JPY instrument so this doesn't pass silently.

Python Decoder (Single Day)

import lzma
import struct
from datetime import datetime, timedelta

def decode_bi5_daily(filepath, day_start, point_value=100000):
    """
    Decode a daily .bi5 tick file.

    :param filepath: Path to the .bi5 file
    :param day_start: datetime representing 00:00 UTC of that day
    :param point_value: 100000 for most pairs, 1000 for JPY pairs
    :return: List of tick dicts
    """
    with open(filepath, 'rb') as f:
        compressed = f.read()

    raw = lzma.decompress(compressed)

    ticks = []
    record_size = 20
    for i in range(0, len(raw), record_size):
        chunk = raw[i:i+record_size]
        ms, ask, bid, ask_vol, bid_vol = struct.unpack('>IIIff', chunk)

        timestamp = day_start + timedelta(milliseconds=ms)
        ticks.append({
            'time': timestamp,
            'ask': ask / point_value,
            'bid': bid / point_value,
            'ask_volume': ask_vol,
            'bid_volume': bid_vol
        })

    return ticks

# Example usage
day_start = datetime(2024, 1, 15, 0, 0, 0)
ticks = decode_bi5_daily('EURUSD/2024/00/15_ticks.bi5', day_start, point_value=100000)

print(f"Total ticks for the day: {len(ticks)}")
for tick in ticks[:5]:
    print(tick)

Batch Decode Script (Full Instrument Folder → CSV)

import lzma
import struct
import csv
import re
import logging
from datetime import datetime, timedelta
from pathlib import Path

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

JPY_PAIRS = {'USDJPY', 'EURJPY', 'GBPJPY', 'AUDJPY', 'NZDJPY', 'CADJPY', 'CHFJPY'}

# Known non-FX instruments (indices/commodities) with a confirmed point value.
# Extend this as you verify additional instruments with the data provider.
OTHER_POINT_VALUES = {
    # 'US30': 100,
    # 'XAUUSD': 100,
}

def get_point_value(instrument):
    instrument = instrument.upper()
    if instrument in JPY_PAIRS:
        return 1000
    if instrument in OTHER_POINT_VALUES:
        return OTHER_POINT_VALUES[instrument]
    if len(instrument) != 6 or not instrument.isalpha():
        logger.warning(
            f"'{instrument}' doesn't look like a standard 6-letter FX pair and has no "
            f"confirmed point value — defaulting to 100000. Verify this before trusting the output."
        )
    return 100000

def decode_bi5_daily(filepath, day_start, point_value):
    with open(filepath, 'rb') as f:
        compressed = f.read()

    raw = lzma.decompress(compressed)

    ticks = []
    record_size = 20
    for i in range(0, len(raw), record_size):
        chunk = raw[i:i+record_size]
        ms, ask, bid, ask_vol, bid_vol = struct.unpack('>IIIff', chunk)
        timestamp = day_start + timedelta(milliseconds=ms)
        ticks.append((timestamp, ask / point_value, bid / point_value, ask_vol, bid_vol))

    return ticks

def batch_decode_instrument(instrument_dir, output_csv):
    """
    Decode all daily .bi5 files for an instrument into a single CSV.
    Expects folder structure: SYMBOL/YEAR/MONTH/DAY_ticks.bi5
    """
    instrument_dir = Path(instrument_dir)
    instrument_name = instrument_dir.name.upper()
    point_value = get_point_value(instrument_name)

    pattern = re.compile(r'(\d{2})_ticks\.bi5$')

    all_ticks = []
    file_count = 0
    error_count = 0

    for year_dir in sorted(instrument_dir.iterdir()):
        if not year_dir.is_dir():
            continue
        for month_dir in sorted(year_dir.iterdir()):
            if not month_dir.is_dir():
                continue

            year = int(year_dir.name)
            month = int(month_dir.name) + 1

            for bi5_file in sorted(month_dir.glob('*_ticks.bi5')):
                match = pattern.search(bi5_file.name)
                if not match:
                    continue

                day = int(match.group(1))
                day_start = datetime(year, month, day)

                try:
                    ticks = decode_bi5_daily(bi5_file, day_start, point_value)
                    all_ticks.extend(ticks)
                    file_count += 1
                except Exception as e:
                    print(f"✗ Error decoding {bi5_file}: {e}")
                    error_count += 1

    all_ticks.sort(key=lambda t: t[0])

    with open(output_csv, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow(['timestamp', 'ask', 'bid', 'ask_volume', 'bid_volume'])
        for tick in all_ticks:
            writer.writerow([tick[0].isoformat(), tick[1], tick[2], tick[3], tick[4]])

    print(f"\n=== Decode Summary ===")
    print(f"Instrument: {instrument_name}")
    print(f"Files processed: {file_count}")
    print(f"Errors: {error_count}")
    print(f"Total ticks: {len(all_ticks)}")
    print(f"Output: {output_csv}")

if __name__ == "__main__":
    batch_decode_instrument('./data/EURUSD', './EURUSD_ticks.csv')

Usage:

python3 decode_bi5_batch.py

Common Pitfalls

• ⚠️ Month is zero-indexed in the S3 path but not in date math — convert explicitly.
• ⚠️ Wrong point value: 100,000 vs 1,000 mismatch silently corrupts prices. Non-FX instruments (indices/commodities) aren't covered by this rule at all — confirm their point value explicitly (see OTHER_POINT_VALUES above).
• ⚠️ Missing files ≠ errors: Treat as "no ticks that day," not a decode failure.
• ⚠️ LZMA raw stream: Some libraries require explicit "raw" mode — standard .xz decompressors will fail.
• ⚠️ Legacy hourly files: Older files may use ms since start of hour instead of ms since start of day. Neither decoder in this guide detects this automatically — both assume the current daily format (SYMBOL/YEAR/MONTH/DAY_ticks.bi5) throughout. If you're working with an instrument or date range old enough to predate the daily layout, verify the file's actual time base before decoding rather than assuming it matches the daily format described here.

End-to-End Pipeline (Download → Decode → Export)

Combines download, decode, and export into one automated script — with optional date filtering and cleanup.

Dependencies

pip install boto3 pandas pyarrow

pyarrow is required only for Parquet export (recommended for large datasets).

Full Pipeline Script

#!/usr/bin/env python3
"""
End-to-End Pipeline: Download -> Decode -> Export
Dukascopy-style .bi5 daily tick files from S3 (Requester Pays)
"""

import argparse
import logging
import lzma
import struct
import shutil
from pathlib import Path
from datetime import datetime, timedelta

import boto3
import pandas as pd
from botocore.config import Config

BUCKET = 'cfg-public-proper-wallaby'
REGION = 'eu-west-1'

JPY_PAIRS = {'USDJPY', 'EURJPY', 'GBPJPY', 'AUDJPY', 'NZDJPY', 'CADJPY', 'CHFJPY'}

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

s3_config = Config(
    max_pool_connections=20,
    retries={'max_attempts': 3, 'mode': 'adaptive'},
    connect_timeout=5,
    read_timeout=60,
    region_name=REGION
)
s3_client = boto3.client('s3', config=s3_config)
s3_resource = boto3.resource('s3', config=s3_config)

def get_point_value(instrument):
    instrument = instrument.upper()
    if instrument in JPY_PAIRS:
        return 1000
    if len(instrument) != 6 or not instrument.isalpha():
        logger.warning(
            f"'{instrument}' doesn't look like a standard 6-letter FX pair — "
            f"defaulting point value to 100000. Verify this before trusting the output."
        )
    return 100000

# Step 1: Download

def download_instrument(instrument, destination, start_date=None, end_date=None):
    """Download all .bi5 files for an instrument, optionally filtered by date range."""
    destination = Path(destination)
    destination.mkdir(parents=True, exist_ok=True)

    bucket_obj = s3_resource.Bucket(BUCKET)
    prefix = f"{instrument}/"

    downloaded = 0
    skipped = 0
    failed = 0

    logger.info(f"Listing objects for {instrument}...")
    objects = list(bucket_obj.objects.filter(Prefix=prefix, RequestPayer='requester'))
    logger.info(f"Found {len(objects)} objects")

    for obj in objects:
        if obj.key.endswith('/'):
            continue

        parts = obj.key.split('/')
        if len(parts) < 4:
            continue

        try:
            year = int(parts[1])
            month = int(parts[2]) + 1
            day = int(parts[3].split('_')[0])
            file_date = datetime(year, month, day)
        except (ValueError, IndexError):
            logger.warning(f"Could not parse date from key: {obj.key}")
            continue

        if start_date and file_date < start_date:
            skipped += 1
            continue
        if end_date and file_date > end_date:
            skipped += 1
            continue

        local_path = destination / parts[1] / parts[2] / parts[3]
        local_path.parent.mkdir(parents=True, exist_ok=True)

        try:
            s3_client.download_file(
                Bucket=BUCKET,
                Key=obj.key,
                Filename=str(local_path),
                ExtraArgs={'RequestPayer': 'requester'}
            )
            downloaded += 1
        except Exception as e:
            logger.error(f"✗ Failed to download {obj.key}: {e}")
            failed += 1

    logger.info(f"Download complete: {downloaded} downloaded, {skipped} skipped (date filter), {failed} failed")
    return downloaded, failed

# Step 2: Decode

def decode_bi5_daily(filepath, day_start, point_value):
    with open(filepath, 'rb') as f:
        compressed = f.read()

    raw = lzma.decompress(compressed)

    ticks = []
    record_size = 20
    for i in range(0, len(raw), record_size):
        chunk = raw[i:i + record_size]
        ms, ask, bid, ask_vol, bid_vol = struct.unpack('>IIIff', chunk)
        timestamp = day_start + timedelta(milliseconds=ms)
        ticks.append((timestamp, ask / point_value, bid / point_value, ask_vol, bid_vol))

    return ticks

def decode_instrument_folder(instrument_dir, instrument_name):
    """Decode all daily .bi5 files in a folder into a list of tick tuples."""
    instrument_dir = Path(instrument_dir)
    point_value = get_point_value(instrument_name)

    all_ticks = []
    file_count = 0
    error_count = 0

    for year_dir in sorted(instrument_dir.iterdir()):
        if not year_dir.is_dir():
            continue
        for month_dir in sorted(year_dir.iterdir()):
            if not month_dir.is_dir():
                continue

            year = int(year_dir.name)
            month = int(month_dir.name) + 1

            for bi5_file in sorted(month_dir.glob('*_ticks.bi5')):
                try:
                    day = int(bi5_file.name.split('_')[0])
                    day_start = datetime(year, month, day)
                    ticks = decode_bi5_daily(bi5_file, day_start, point_value)
                    all_ticks.extend(ticks)
                    file_count += 1
                except Exception as e:
                    logger.error(f"✗ Error decoding {bi5_file}: {e}")
                    error_count += 1

    all_ticks.sort(key=lambda t: t[0])
    logger.info(f"Decoded {file_count} files ({error_count} errors), {len(all_ticks)} total ticks")

    return all_ticks

# Step 3: Export

def export_ticks(ticks, output_path, output_format='csv'):
    """Export decoded ticks to CSV or Parquet using pandas."""
    df = pd.DataFrame(ticks, columns=['timestamp', 'ask', 'bid', 'ask_volume', 'bid_volume'])

    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    if output_format == 'csv':
        df.to_csv(output_path, index=False)
    elif output_format == 'parquet':
        df.to_parquet(output_path, index=False, engine='pyarrow', compression='snappy')
    else:
        raise ValueError(f"Unsupported format: {output_format}")

    logger.info(f"✓ Exported {len(df)} ticks to {output_path} ({output_format})")
    return df

# Pipeline Orchestration

def run_pipeline(instrument, start_date=None, end_date=None,
                  output_format='parquet', cleanup=False,
                  raw_dir=None, output_dir='./output'):

    instrument = instrument.upper()
    raw_dir = raw_dir or f"./raw/{instrument}"
    output_path = Path(output_dir) / f"{instrument}_ticks.{output_format}"

    logger.info(f"=== Pipeline started for {instrument} ===")
    if start_date or end_date:
        logger.info(f"Date range: {start_date or 'earliest'} to {end_date or 'latest'}")

    downloaded, failed = download_instrument(instrument, raw_dir, start_date, end_date)
    if failed:
        logger.warning(f"{failed} file(s) failed to download for {instrument} — proceeding with the {downloaded} that succeeded")
    if downloaded == 0:
        logger.warning(f"No files downloaded for {instrument} — skipping decode/export")
        return

    ticks = decode_instrument_folder(raw_dir, instrument)
    if not ticks:
        logger.warning(f"No ticks decoded for {instrument}")
        return

    export_ticks(ticks, output_path, output_format)

    if cleanup:
        logger.info(f"Cleaning up raw .bi5 files at {raw_dir}...")
        shutil.rmtree(raw_dir, ignore_errors=True)
        logger.info("✓ Cleanup complete")

    logger.info(f"=== Pipeline finished for {instrument} ===\n")

def parse_date(date_str):
    return datetime.strptime(date_str, '%Y-%m-%d')

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Download, decode, and export Dukascopy .bi5 tick data")
    parser.add_argument('instrument', help="Currency pair, e.g. EURUSD")
    parser.add_argument('--start-date', type=parse_date, help="YYYY-MM-DD")
    parser.add_argument('--end-date', type=parse_date, help="YYYY-MM-DD")
    parser.add_argument('--format', choices=['csv', 'parquet'], default='parquet')
    parser.add_argument('--cleanup', action='store_true', help="Delete raw .bi5 files after export")
    parser.add_argument('--output-dir', default='./output')
    parser.add_argument('--raw-dir', default=None, help="Where to store raw .bi5 files (default: ./raw/<INSTRUMENT>)")

    args = parser.parse_args()

    run_pipeline(
        instrument=args.instrument,
        start_date=args.start_date,
        end_date=args.end_date,
        output_format=args.format,
        cleanup=args.cleanup,
        raw_dir=args.raw_dir,
        output_dir=args.output_dir
    )

Usage Examples

# Download, decode, and export EURUSD to Parquet (default)
python3 pipeline.py EURUSD

# Export to CSV instead
python3 pipeline.py EURUSD --format csv

# Filter to a specific date range
python3 pipeline.py EURUSD --start-date 2024-01-01 --end-date 2024-03-31

# Clean up raw .bi5 files after export
python3 pipeline.py EURUSD --cleanup

# Custom output directory
python3 pipeline.py GBPUSD --output-dir ./exports --cleanup

Batch Pipeline for Multiple Instruments

#!/bin/bash
# run_all_pipelines.sh

INSTRUMENTS=("EURUSD" "GBPUSD" "USDJPY" "AUDUSD")

for INSTRUMENT in "${INSTRUMENTS[@]}"; do
    echo "=== Processing $INSTRUMENT ==="
    python3 pipeline.py "$INSTRUMENT" --format parquet --cleanup
done

echo "✓ All pipelines complete"

Best Practices ✅

Download & Cost Control

• ✓ Always include --region eu-west-1 and --request-payer requester flags
• ✓ Verify actual size/file count per pair with --summarize before bulk downloads — averages can mislead (e.g., EUR/USD's real average file size is ~5x the archive-wide average)
• ✓ Use aws s3 sync (not cp) for multiple files — only transfers changed data
• ✓ Use --delete flag to avoid re-downloading existing files
• ✓ Run instrument discovery first to confirm exact, valid pair names — avoids failed syncs from typos
• ✓ Keep log files for audit trails
• ✓ Set up CloudWatch alarms for failed downloads
• ✓ Recalculate costs whenever bucket size/file count changes materially
• ✓ Prefer downloading specific pairs over the full archive when possible

Decoding & Data Integrity

• ✓ Always confirm the correct point value (100,000 vs 1,000) per instrument before decoding — wrong scaling silently corrupts prices
• ✓ Treat missing daily .bi5 files as "no ticks that day" (weekends/holidays), not as errors
• ✓ Never mix legacy hourly .bi5 files with new daily files in the same pipeline without detecting format first
• ✓ Validate LZMA decompression mode — some libraries require explicit "raw" mode (no container headers)
• ✓ Sort decoded ticks chronologically before export, in case of filesystem ordering inconsistencies

Pipeline & Automation

• ✓ Use --cleanup to delete raw .bi5 files after decoding — they can always be re-downloaded from S3 if needed
• ✓ Prefer Parquet over CSV for tick-level data — typically 5-10x smaller and faster to query
• ✓ Use --start-date / --end-date filters when testing, to avoid unnecessary S3 costs
• ✓ Combine with instrument discovery to loop over all valid pairs automatically instead of hardcoding lists
• ✓ Log every pipeline run (downloaded/decoded/exported counts) for auditability

General

• ✓ Run downloads during off-peak hours to reduce contention (not cost)
• ✓ Implement checksums for data integrity verification where critical
• ✓ Monitor total transferred data over time to control cumulative costs