File size: 3,340 Bytes
2bbc43c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | #!/usr/bin/env python3
"""Convert ALL downloaded Binance ZIP files to parquet in one batch.
Usage:
python scripts/convert_all_binance.py --input data/ --output data/parquet/
"""
import argparse
import io
import logging
import zipfile
from pathlib import Path
import pandas as pd
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)
def convert_zip(zip_path: Path, output_path: Path) -> bool:
"""Convert one Binance trades ZIP to parquet. Returns True on success."""
if output_path.exists():
logger.info(" SKIP (exists): %s", output_path.name)
return True
try:
with zipfile.ZipFile(zip_path, "r") as zf:
csv_name = zf.namelist()[0]
with zf.open(csv_name) as f:
df = pd.read_csv(
io.BytesIO(f.read()),
header=None,
names=["id", "price", "qty", "quoteQty", "time", "isBuyerMaker", "isBestMatch"],
)
df_out = pd.DataFrame({
"timestamp_ms": df["time"].astype("int64"),
"best_bid": df["price"] * 0.9999,
"best_ask": df["price"] * 1.0001,
"bid_size": 1.0,
"ask_size": 1.0,
"mid_price": df["price"].astype("float64"),
"trade_price": df["price"].astype("float64"),
"trade_size": df["qty"].astype("float64"),
"trade_side": df["isBuyerMaker"].map({True: "sell", False: "buy"}),
}).sort_values("timestamp_ms").reset_index(drop=True)
output_path.parent.mkdir(parents=True, exist_ok=True)
df_out.to_parquet(output_path, index=False)
size_mb = output_path.stat().st_size / (1024 * 1024)
logger.info(" ✓ %s -> %s (%.1f MB, %d ticks)",
zip_path.name, output_path.name, size_mb, len(df_out))
return True
except Exception as e:
logger.error(" ✗ %s: %s", zip_path.name, e)
return False
def main() -> int:
parser = argparse.ArgumentParser(description="Convert all Binance ZIPs to parquet")
parser.add_argument("--input", default="data/", help="Directory with .zip files")
parser.add_argument("--output", default="data/parquet/", help="Output directory")
args = parser.parse_args()
input_dir = Path(args.input)
output_dir = Path(args.output)
zips = sorted(input_dir.glob("*-trades-*.zip"))
if not zips:
logger.error("No *-trades-*.zip files found in %s", input_dir)
return 1
logger.info("Found %d ZIP files to convert", len(zips))
success = 0
for zip_path in zips:
# e.g., BTCUSDT-trades-2021-05-19.zip -> BTCUSDT_2021-05-19.parquet
stem = zip_path.stem # BTCUSDT-trades-2021-05-19
parts = stem.split("-", 1) # ["BTCUSDT", "trades-2021-05-19"]
symbol = parts[0]
date_part = parts[1].replace("trades-", "") # 2021-05-19
out_name = f"{symbol}_{date_part}.parquet"
out_path = output_dir / out_name
if convert_zip(zip_path, out_path):
success += 1
logger.info("=" * 50)
logger.info("Converted %d/%d files", success, len(zips))
logger.info("Output: %s", output_dir.resolve())
return 0 if success else 1
if __name__ == "__main__":
raise SystemExit(main())
|