#!/usr/bin/env python3
"""
Process collected Bitcoin documentation into training format for fine-tuning.
Converts raw documentation into instruction-response pairs optimized for
a Bitcoin operations assistant.
"""
import json
import re
from pathlib import Path
from typing import List, Dict
class BitcoinTrainingProcessor:
def __init__(self, data_dir="bitcoin_data"):
# Fix: Use the actual data directory path
self.data_dir = Path(__file__).parent / data_dir
self.training_data = []
def extract_sections(self, text: str, category: str) -> List[Dict]:
"""Extract logical sections from documentation."""
sections = []
# Split by headers (markdown style)
headers = re.split(r'\n(?=#\s)', text)
for header in headers:
if len(header.strip()) < 50: # Skip very short sections
continue
# Extract title and content
lines = header.split('\n', 1)
if len(lines) == 2:
title = lines[0].replace('#', '').strip()
content = lines[1].strip()
else:
title = "Bitcoin Operations Guide"
content = header.strip()
if len(content) > 100: # Ensure substantial content
sections.append({
"title": title,
"content": content,
"category": category
})
return sections
def create_instruction_response(self, section: Dict) -> Dict:
"""Create an instruction-response pair from a documentation section.
Uses conversational tone like talking to a knowledgeable friend.
"""
title = section['title']
content = section['content']
category = section['category']
# Create diverse instruction formats with conversational tone
instructions = [
f"How would you explain {title} to someone who knows Bitcoin basics?",
f"I'm trying to understand {title} - can you walk me through it?",
f"Hey, I need to set up {title} - what's the command line way to do that?",
f"What's the deal with {title} in Bitcoin? Explain it like I'm a friend.",
f"Can you show me the CLI commands for {title}?",
f"I'm stuck with {title} - what should I check?",
f"How do I troubleshoot {title} issues?",
f"What are the best practices for {title}?",
f"Walk me through setting up {title} step by step",
f"What commands would you run to work with {title}?",
f"I want to understand {title} from a technical perspective",
f"Can you explain {title} like we're having a beer and talking Bitcoin?"
]
# Select instruction based on content type and category
instruction_idx = len(self.training_data) % len(instructions)
instruction = instructions[instruction_idx]
return {
"instruction": instruction,
"input": "",
"output": content,
"category": category
}
def process_all_docs(self):
"""Process all collected documentation."""
print("Processing collected Bitcoin documentation...")
categories = [
"bips",
"lightning_protocol",
"lnd",
"cln",
"bitcoin_core",
"mining",
"wallet_security",
"scripting",
"real_time",
"community",
"security",
"privacy",
"lightning_plugins",
"troubleshooting"
]
for category in categories:
category_dir = self.data_dir / category
if category_dir.exists():
print(f"Processing {category}...")
for doc_file in category_dir.glob("*.md"):
with open(doc_file, 'r', encoding='utf-8') as f:
content = f.read()
# Extract sections and create training pairs
sections = self.extract_sections(content, category)
for section in sections:
training_pair = self.create_instruction_response(section)
self.training_data.append(training_pair)
print(f" Processed {len(list(category_dir.glob('*.md')))} documents")
# Handle JSON files (troubleshooting scenarios, real-time data)
if category_dir.exists():
for json_file in category_dir.glob("*.json"):
with open(json_file, 'r', encoding='utf-8') as f:
try:
data = json.load(f)
if isinstance(data, list):
for item in data:
# Convert JSON data to instruction-response format
instruction = f"Here's some {category} data - what does it mean?"
response = json.dumps(item, indent=2)
self.training_data.append({
"instruction": instruction,
"input": "",
"output": response,
"category": category
})
except json.JSONDecodeError:
print(f" Skipping invalid JSON: {json_file}")
print(f"Total training pairs created: {len(self.training_data)}")
def save_training_data(self, output_path="bitcoin_finetuning.json"):
"""Save processed training data."""
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(self.training_data, f, indent=2, ensure_ascii=False)
print(f"Training data saved to {output_path}")
def process_and_save(self):
"""Process all docs and save training data."""
self.process_all_docs()
self.save_training_data()
if __name__ == "__main__":
processor = BitcoinTrainingProcessor()
processor.process_and_save()