#!/usr/bin/env python3 """ Data Streaming and Parsing Utility for the HackIndia Challenge. Author: Team Ascended Description: Memory-efficient streaming of financial QA datasets (specifically mishface123/adaption-econ-finance-qa-pairs) with batch generation, parsing, and formatting capabilities. """ import argparse import sys from typing import Dict, Generator, List, Optional try: from datasets import load_dataset except ImportError: print("Warning: 'datasets' library not found. Please install it via 'pip install datasets'.") def stream_dataset( dataset_name: str, split: str = "train", batch_size: int = 1, limit: Optional[int] = None ) -> Generator[List[Dict], None, None]: """ Streams a dataset from Hugging Face and yields parsed data in batches. Args: dataset_name: Hugging Face dataset identifier. split: The dataset split to stream (e.g., 'train', 'test'). batch_size: Number of records to yield at a time. limit: Max number of records to process before stopping. Yields: A list of parsed dictionaries containing formatted instruction-response pairs. """ try: # Load the dataset in streaming mode to minimize RAM overhead dataset = load_dataset(dataset_name, split=split, streaming=True) except Exception as e: print(f"Error loading dataset {dataset_name}: {e}", file=sys.stderr) return batch = [] count = 0 for record in dataset: parsed_record = parse_record(record) batch.append(parsed_record) count += 1 if len(batch) == batch_size: yield batch batch = [] if limit is not None and count >= limit: break if batch: yield batch def parse_record(record: Dict) -> Dict: """ Parses a single record from the dataset and standardizes its keys. Designed specifically for 'mishface123/adaption-econ-finance-qa-pairs'. Args: record: The raw record dict from the dataset. Returns: A standardized dictionary containing instruction, context, and response. """ # Standardize keys depending on schema instruction = record.get("instruction") or record.get("question") or "" context = record.get("context") or "" response = record.get("output") or record.get("response") or record.get("answer") or "" metadata = record.get("metadata") or {} # Extract target market metadata (e.g., Indian localization indicator) localized = metadata.get("localized", False) or "india" in str(metadata).lower() return { "instruction": instruction.strip(), "context": context.strip(), "response": response.strip(), "localized": localized, "raw_keys": list(record.keys()) } def main(): parser = argparse.ArgumentParser( description="Stream and parse datasets for the AutoScientist pipeline." ) parser.add_argument( "--dataset", type=str, default="mishface123/adaption-econ-finance-qa-pairs", help="Hugging Face dataset name to stream." ) parser.add_argument( "--split", type=str, default="train", help="Dataset split (train, validation, test)." ) parser.add_argument( "--batch-size", type=int, default=4, help="Batch size for yield/printing." ) parser.add_argument( "--limit", type=int, default=10, help="Maximum records to stream/parse." ) args = parser.parse_args() print(f"Streaming dataset: {args.dataset} (split: {args.split})...") batch_generator = stream_dataset( dataset_name=args.dataset, split=args.split, batch_size=args.batch_size, limit=args.limit ) for i, batch in enumerate(batch_generator): print(f"\n--- Batch {i+1} (Size: {len(batch)}) ---") for record_idx, record in enumerate(batch): print(f"\nRecord {record_idx+1}:") print(f" Instruction : {record['instruction'][:120]}...") print(f" Context : {record['context'][:120]}...") print(f" Response : {record['response'][:120]}...") print(f" Localized : {record['localized']}") if __name__ == "__main__": main()