import os import json import glob import re import hashlib from datetime import datetime, timezone import pandas as pd import gradio as gr import spaces from pypdf import PdfReader import docx2txt from llama_index.core.node_parser import SentenceSplitter # Optional image/OCR dependencies are imported lazily so the Space can still # start when a specific vision backend is unavailable. try: import pytesseract from PIL import Image except Exception: pytesseract = None Image = None # ============================================================= # TRADING LLM DATASET CREATOR # ============================================================= # Two logically separate knowledge domains: # # 1) SYSTEM RAG / ADMIN KNOWLEDGE (private, not user-downloadable) # /data/system_knowledge_base # Contains the dataset-generation rules, schemas, strategy-library rules, # timestamp conventions, image-to-text rules, and Unsloth formatting rules. # # 2) USER DATA WORKSPACE # /data/user_workspace # Contains only the files uploaded by the current dataset-building workflow. # # The RAG layer is intentionally local and deterministic. It retrieves the # relevant private rules before a dataset is generated. This prevents the # application from blindly applying one generic template to every trading # dataset. # ============================================================= SYSTEM_KB_DIR = "/data/system_knowledge_base" USER_ROOT = "/data/user_workspace" UPLOAD_DIR = os.path.join(USER_ROOT, "raw_inputs") PROCESSED_DIR = os.path.join(USER_ROOT, "processed_jsonl") MASTER_FILE = os.path.join(USER_ROOT, "master_dataset.jsonl") RAG_INDEX_FILE = os.path.join(SYSTEM_KB_DIR, "rag_index.jsonl") for directory in [SYSTEM_KB_DIR, UPLOAD_DIR, PROCESSED_DIR]: os.makedirs(directory, exist_ok=True) SUPPORTED_EXTENSIONS = [ ".pdf", ".txt", ".md", ".docx", ".csv", ".xlsx", ".png", ".jpg", ".jpeg", ".webp" ] # ------------------------------------------------------------- # PRIVATE BACKEND RAG KNOWLEDGE BASE # ------------------------------------------------------------- DEFAULT_SYSTEM_RULES = { "dataset_generation_rules.md": """ TRADING LLM DATASET GENERATION RULES The dataset creator must distinguish between: A. Strategy knowledge / documentation datasets. B. Numerical market-data datasets. C. Image/chart understanding datasets. D. Conversational trading-analysis datasets. E. Trade-decision approval or denial datasets. Never mix these formats without explicitly labeling the source type and target training objective. A strategy library should preserve semantic structure. A PDF containing a strategy, rulebook, risk model, or trading methodology should be extracted, cleaned, separated into coherent chunks, and represented with metadata such as document name, section, page when available, topic, strategy name, timeframe, market, and source type. A numerical dataset must preserve numerical values exactly. Do not replace OHLCV or indicator values with vague natural-language summaries when the target model is expected to learn quantitative relationships. Preserve timestamps, symbol, timeframe, OHLCV, indicator columns, and labels. Rows must be sorted chronologically within each symbol/timeframe series. If timestamps are missing, the system must not silently invent real-world dates that could be mistaken for historical market data. A generated sequence must be explicitly marked as synthetic/generated and must preserve row order. Every training example should have a clearly defined target objective. Examples: market-state classification, setup detection, trade-plan generation, trade approval/denial, chart captioning, strategy explanation, or multimodal chart analysis. Avoid look-ahead leakage. Features used at decision time must not contain future information. If a label uses future candles, the label horizon and future window must be explicitly documented in metadata. The system should preserve provenance. Each output row should identify the source file and dataset type. """, "strategy_library_rules.md": """ STRATEGY LIBRARY DATASET RULES Strategy documents are knowledge assets, not ordinary prose blobs. The extraction pipeline should preserve: - strategy name - market/instrument - timeframe - session - setup conditions - entry conditions - invalidation conditions - stop-loss logic - take-profit logic - position sizing/risk rules - confirmation requirements - no-trade conditions - examples and counterexamples When a document contains rules, the dataset should not invent missing rules. When a rule is ambiguous, preserve the ambiguity in the source text or flag it for review. A strategy-library knowledge-base row should be suitable for retrieval and should contain a coherent context chunk rather than an arbitrary cut through a table or rule. For conversational fine-tuning, use instruction/input/response or the target chat format selected by the user. Do not fabricate a profitable trade outcome. """, "market_data_rules.md": """ NUMERICAL MARKET DATA RULES Numerical CSV/XLSX data should be treated as structured time-series data. Recommended metadata: symbol, timeframe, timestamp, source_file, row_id, dataset_type, feature_columns, label_columns, synthetic_timestamp. Recognize common columns case-insensitively: timestamp/date/datetime/time, symbol/ticker, open, high, low, close, volume, vwap, atr, rsi, macd, ema, sma, adx, bbands, and other indicators. Do not round values unless the user explicitly requests rounding. Do not convert numeric data into prose as the only representation. For numerical fine-tuning, retain structured fields and optionally add a natural-language market_state field. Rows must be sorted by timestamp. Duplicate timestamps should be preserved only when the dataset has a valid reason, such as multiple symbols or levels. A generated timestamp must be labeled synthetic_timestamp=true. It is a sequence index, not evidence of a real historical date. The dataset creator should support labels such as: setup_present, setup_type, market_regime, direction, entry, stop_loss, take_profit, risk_reward, outcome, approval, denial_reason, and confidence. Labels must come from the source data or user-provided annotations. """, "image_chart_rules.md": """ TRADING IMAGE AND CHART DATASET RULES Chart screenshots and candlestick images can be used for: 1. image-to-text captioning, 2. visual question answering, 3. chart-analysis conversations, 4. multimodal supervised fine-tuning. Do not claim exact price values from a chart image unless they are legible or provided as structured metadata. Do not infer hidden candles or indicators. A chart-image example should preserve: image path or image identifier, symbol if known, timeframe if known, visible chart context, visual observations, market structure, liquidity observations, indicators visible, setup classification, entry/stop/target only when provided or clearly inferable, and uncertainty notes. For a multimodal dataset, the image reference must remain associated with the text response. For a text-only dataset, OCR and/or a vision caption may be used, but the generated text must be marked as image-derived. A chart-analysis response should distinguish observation from inference and should not manufacture certainty. """, "unsloth_formats.md": """ UNSLOTH DATASET FORMAT RULES The dataset creator should support multiple output schemas rather than forcing all data into one universal format. Supported logical dataset types: - strategy_knowledge - market_time_series - chart_image_caption - chart_analysis_conversation - trade_decision - mixed_multimodal Preferred instruction dataset fields: instruction, input, output, metadata Preferred conversational field: conversations: [{"from":"system","value":"..."},{"from":"human","value":"..."},{"from":"gpt","value":"..."}] For multimodal data, retain an image field or image reference according to the training pipeline selected by the user. Each row should contain provenance metadata whenever possible. Dataset rows must be valid JSONL and independently parseable. The creator should not promise that a dataset is automatically suitable for every Unsloth model. The selected base model, tokenizer, chat template, sequence length, and trainer configuration must match the generated schema. """, } def ensure_default_system_knowledge(): """Create starter private rules only when the backend KB is empty.""" existing = [ p for p in glob.glob(os.path.join(SYSTEM_KB_DIR, "*")) if os.path.isfile(p) and os.path.basename(p) != "rag_index.jsonl" ] if existing: return for filename, text in DEFAULT_SYSTEM_RULES.items(): with open(os.path.join(SYSTEM_KB_DIR, filename), "w", encoding="utf-8") as f: f.write(text.strip() + "\n") def normalize_text(text): text = text or "" text = text.replace("\x00", " ") text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\n{3,}", "\n\n", text) return text.strip() def tokenize_for_rag(text): return set(re.findall(r"[a-zA-Z0-9_]{3,}", text.lower())) def build_private_rag_index(): """ Lightweight lexical RAG index. It intentionally keeps the private system rules on the backend and never exposes them through the user download. """ ensure_default_system_knowledge() splitter = SentenceSplitter(chunk_size=700, chunk_overlap=100) records = [] for path in sorted(glob.glob(os.path.join(SYSTEM_KB_DIR, "*"))): if not os.path.isfile(path) or os.path.basename(path) == "rag_index.jsonl": continue ext = os.path.splitext(path)[1].lower() if ext not in [".txt", ".md", ".pdf", ".docx"]: continue try: text = parse_text_document(path) chunks = splitter.split_text(normalize_text(text)) for i, chunk in enumerate(chunks): records.append({ "source": os.path.basename(path), "chunk_id": i, "text": chunk, "terms": sorted(tokenize_for_rag(chunk)), }) except Exception: continue with open(RAG_INDEX_FILE, "w", encoding="utf-8") as f: for row in records: f.write(json.dumps(row, ensure_ascii=False) + "\n") return records def retrieve_private_rules(query, top_k=6): if not os.path.exists(RAG_INDEX_FILE): records = build_private_rag_index() else: records = [] with open(RAG_INDEX_FILE, "r", encoding="utf-8") as f: for line in f: try: records.append(json.loads(line)) except Exception: pass q_terms = tokenize_for_rag(query) scored = [] for row in records: terms = set(row.get("terms", [])) overlap = len(q_terms & terms) phrase_bonus = sum(1 for term in q_terms if term in row.get("text", "").lower()) score = overlap * 3 + phrase_bonus if score > 0: scored.append((score, row)) scored.sort(key=lambda x: x[0], reverse=True) return [row for _, row in scored[:top_k]] # ------------------------------------------------------------- # EXTRACTION # ------------------------------------------------------------- def parse_text_document(file_path): ext = os.path.splitext(file_path)[1].lower() text = "" if ext in [".txt", ".md"]: with open(file_path, "r", encoding="utf-8", errors="ignore") as f: text = f.read() elif ext == ".pdf": reader = PdfReader(file_path) pages = [] for page_no, page in enumerate(reader.pages, start=1): page_text = page.extract_text() or "" if page_text.strip(): pages.append(f"[PAGE {page_no}]\n{page_text}") text = "\n\n".join(pages) elif ext == ".docx": text = docx2txt.process(file_path) return normalize_text(text) def infer_column(df, candidates): lookup = {str(c).strip().lower(): c for c in df.columns} for candidate in candidates: if candidate in lookup: return lookup[candidate] return None def detect_indicator_columns(df): known = { "vwap", "rvol", "atr", "rsi", "macd", "adx", "ema", "sma", "bbands", "bollinger", "stoch", "obv", "cci", "supertrend" } return [ c for c in df.columns if str(c).strip().lower() in known or any(token in str(c).lower() for token in known) ] def process_market_dataframe(df, file_name, symbol_hint="", timeframe_hint=""): """ Preserves structured numeric data and adds explicit metadata. It does not convert quantitative rows into prose as the primary representation. """ df = df.copy() df.columns = [str(c).strip() for c in df.columns] time_col = infer_column(df, [ "timestamp", "datetime", "date", "time", "bar_time", "candle_time" ]) symbol_col = infer_column(df, ["symbol", "ticker", "instrument", "asset"]) timeframe_col = infer_column(df, ["timeframe", "interval", "tf"]) synthetic_timestamp = False if time_col: parsed = pd.to_datetime(df[time_col], errors="coerce", utc=True) valid = parsed.notna() df = df.loc[valid].copy() df[time_col] = parsed.loc[valid] df = df.sort_values(time_col, kind="stable") else: synthetic_timestamp = True # Sequence index is deliberately not presented as historical market time. df["synthetic_timestamp"] = range(len(df)) time_col = "synthetic_timestamp" lower_cols = {str(c).lower() for c in df.columns} is_ohlcv = bool({"open", "high", "low", "close"} & lower_cols) rows = [] for row_id, (_, row) in enumerate(df.iterrows()): record = {} for col in df.columns: value = row[col] if pd.isna(value): record[col] = None elif isinstance(value, pd.Timestamp): record[col] = value.isoformat() elif hasattr(value, "item"): try: record[col] = value.item() except Exception: record[col] = str(value) else: record[col] = value metadata = { "source_file": file_name, "row_id": row_id, "dataset_type": "market_time_series" if is_ohlcv else "structured_numeric", "synthetic_timestamp": synthetic_timestamp, "symbol": str(row[symbol_col]) if symbol_col and pd.notna(row[symbol_col]) else symbol_hint or None, "timeframe": str(row[timeframe_col]) if timeframe_col and pd.notna(row[timeframe_col]) else timeframe_hint or None, "indicator_columns": [str(c) for c in detect_indicator_columns(df)], } rows.append({ "data": record, "metadata": metadata }) return rows def extract_image_text(file_path): if Image is None or pytesseract is None: return "", "OCR unavailable: install Pillow and pytesseract/Tesseract for OCR." try: image = Image.open(file_path) text = pytesseract.image_to_string(image) return normalize_text(text), "OCR extracted text from image." except Exception as exc: return "", f"OCR error: {exc}" def parse_user_asset(file_path, symbol_hint="", timeframe_hint=""): ext = os.path.splitext(file_path)[1].lower() name = os.path.basename(file_path) if ext in [".txt", ".md", ".pdf", ".docx"]: text = parse_text_document(file_path) return [{ "record_type": "document", "text": text, "metadata": { "source_file": name, "dataset_type": "strategy_knowledge", "source_extension": ext, } }] if text else [] if ext in [".csv", ".xlsx"]: df = pd.read_csv(file_path) if ext == ".csv" else pd.read_excel(file_path) return process_market_dataframe(df, name, symbol_hint, timeframe_hint) if ext in [".png", ".jpg", ".jpeg", ".webp"]: ocr_text, ocr_status = extract_image_text(file_path) return [{ "record_type": "image", "image_path": file_path, "text": ocr_text, "metadata": { "source_file": name, "dataset_type": "chart_image_caption", "image_derived_text": True, "ocr_status": ocr_status, "symbol": symbol_hint or None, "timeframe": timeframe_hint or None, } }] return [] # ------------------------------------------------------------- # DATASET SCHEMAS # ------------------------------------------------------------- def make_strategy_rows(records, chunk_size, chunk_overlap): splitter = SentenceSplitter(chunk_size=int(chunk_size), chunk_overlap=int(chunk_overlap)) output = [] for record in records: if record.get("record_type") != "document": continue source_file = record["metadata"]["source_file"] chunks = splitter.split_text(record.get("text", "")) for idx, chunk in enumerate(chunks): output.append({ "instruction": "Retrieve and explain the relevant trading strategy knowledge without inventing rules.", "input": chunk, "output": chunk, "metadata": { **record["metadata"], "chunk_id": idx, "source_type": "strategy_document", } }) return output def make_market_rows(records): return [{ "instruction": "Interpret the structured market state while preserving the numerical values and timestamp context.", "input": json.dumps(record["data"], ensure_ascii=False), "output": json.dumps({ "market_state": "structured_market_observation", "data": record["data"] }, ensure_ascii=False), "metadata": record["metadata"] } for record in records if record.get("record_type") is None and "data" in record] def make_image_rows(records, image_prompt): output = [] for record in records: if record.get("record_type") != "image": continue output.append({ "instruction": image_prompt, "input": { "image": record["image_path"], "ocr_text": record.get("text", "") }, "output": ( "Image-derived chart observations:\n" + (record.get("text") or "[No reliable OCR text extracted.]") ), "metadata": record["metadata"] }) return output def make_trade_decision_rows(records, decision_prompt): """ Converts already-labeled source records into an approval/denial style dataset. It does not invent labels. This schema is intended for datasets where the user has supplied the decision or decision labels. """ output = [] for record in records: if record.get("record_type") == "document": output.append({ "instruction": decision_prompt, "input": record.get("text", ""), "output": "REVIEW_REQUIRED: No explicit trade decision label was provided.", "metadata": { **record["metadata"], "label_status": "missing", } }) return output # ------------------------------------------------------------- # MAIN PIPELINE # ------------------------------------------------------------- def execute_trading_dataset_pipeline( files, dataset_type, enable_chunking, chunk_size, chunk_overlap, symbol_hint, timeframe_hint, image_prompt, decision_prompt, ): if not files: return "⚠️ Upload at least one source file.", None, "" os.makedirs(PROCESSED_DIR, exist_ok=True) for path in glob.glob(os.path.join(PROCESSED_DIR, "*.jsonl")): try: os.remove(path) except Exception: pass # Retrieve private backend rules before transforming user data. rag_query = f""" dataset type: {dataset_type} symbol: {symbol_hint} timeframe: {timeframe_hint} image prompt: {image_prompt} trade decision: {decision_prompt} """ rules = retrieve_private_rules(rag_query) rule_sources = sorted(set(r["source"] for r in rules)) all_records = [] processed = 0 errors = [] for file_obj in files: try: path = file_obj.name if hasattr(file_obj, "name") else str(file_obj) records = parse_user_asset(path, symbol_hint, timeframe_hint) if records: all_records.extend(records) processed += 1 except Exception as exc: errors.append(f"{os.path.basename(str(file_obj))}: {exc}") if not all_records: return "❌ No supported content could be extracted.", None, "\n".join(errors) if dataset_type == "Strategy Knowledge Base": rows = make_strategy_rows(all_records, chunk_size, chunk_overlap) elif dataset_type == "Market Time-Series / XGBoost": rows = make_market_rows(all_records) elif dataset_type == "Chart Image → Text": rows = make_image_rows(all_records, image_prompt) elif dataset_type == "Trade Approval / Denial": rows = make_trade_decision_rows(all_records, decision_prompt) else: # Mixed mode: preserve each logical modality instead of flattening # everything into one lossy text representation. rows = ( make_strategy_rows(all_records, chunk_size, chunk_overlap) + make_market_rows(all_records) + make_image_rows(all_records, image_prompt) ) if not rows: return "❌ Files were extracted, but no rows matched the selected dataset type.", None, "" run_metadata = { "creator": "Trading LLM Dataset Creator", "created_at_utc": datetime.now(timezone.utc).isoformat(), "dataset_type": dataset_type, "processed_files": processed, "rows": len(rows), "private_rag_rule_sources": rule_sources, "symbol_hint": symbol_hint or None, "timeframe_hint": timeframe_hint or None, "schema_version": "trading-llm-dataset-v1", } with open(MASTER_FILE, "w", encoding="utf-8") as f: for row in rows: row["_dataset_metadata"] = run_metadata f.write(json.dumps(row, ensure_ascii=False) + "\n") log = ( "✅ TRADING DATASET BUILD COMPLETE\n\n" f"• Dataset type: {dataset_type}\n" f"• Files processed: {processed}/{len(files)}\n" f"• Rows generated: {len(rows)}\n" f"• Private RAG rules applied: {len(rule_sources)} source documents\n" f"• Backend output: {MASTER_FILE}\n" ) if errors: log += "\n⚠️ Errors:\n" + "\n".join(errors) rag_preview = "\n".join( f"• {r['source']} — chunk {r['chunk_id']}" for r in rules ) return log, MASTER_FILE, rag_preview # ------------------------------------------------------------- # ADMIN-ONLY BACKEND KNOWLEDGE INGESTION # ------------------------------------------------------------- def rebuild_backend_rag_index(admin_files): if not admin_files: return "No backend rule documents supplied." imported = 0 for file_obj in admin_files: path = file_obj.name if hasattr(file_obj, "name") else str(file_obj) ext = os.path.splitext(path)[1].lower() if ext not in [".txt", ".md", ".pdf", ".docx"]: continue target_name = hashlib.sha256( os.path.basename(path).encode("utf-8") ).hexdigest()[:16] + "_" + os.path.basename(path) target = os.path.join(SYSTEM_KB_DIR, target_name) with open(path, "rb") as src, open(target, "wb") as dst: dst.write(src.read()) imported += 1 records = build_private_rag_index() return ( "🔒 PRIVATE BACKEND RAG UPDATED\n\n" f"• Documents imported: {imported}\n" f"• Indexed chunks: {len(records)}\n" "• These rules are not included in user dataset downloads." ) # ------------------------------------------------------------- # UI # ------------------------------------------------------------- custom_theme = gr.themes.Default( primary_hue="green", secondary_hue="zinc", neutral_hue="zinc" ) dataset_types = [ "Strategy Knowledge Base", "Market Time-Series / XGBoost", "Chart Image → Text", "Trade Approval / Denial", "Mixed Multimodal Trading Dataset", ] with gr.Blocks(theme=custom_theme, title="Trading LLM Dataset Creator") as demo: gr.Markdown( """ # 📈 TRADING LLM DATASET CREATOR Create structured datasets for trading-focused LLM fine-tuning and quantitative model workflows. The application separates strategy knowledge, numerical time-series data, chart images, and trade-decision datasets instead of forcing all sources into one generic text format. **Private backend RAG:** the dataset-generation rules live in a backend-only knowledge base and are retrieved before the user dataset is generated. """ ) with gr.Tab("🧠 Create Trading Dataset"): with gr.Row(): with gr.Column(scale=1): file_uploader = gr.File( file_count="multiple", type="filepath", label="📥 Upload PDFs, TXT, MD, DOCX, CSV, XLSX, PNG, JPG, WEBP" ) dataset_type = gr.Dropdown( choices=dataset_types, value=dataset_types[0], label="🎯 Target Dataset Type" ) with gr.Row(): symbol_input = gr.Textbox( label="Symbol / Instrument (optional)", placeholder="EURUSD, BTCUSDT, ES, AAPL..." ) timeframe_input = gr.Textbox( label="Timeframe (optional)", placeholder="1m, 5m, 1H, 4H, 1D..." ) with gr.Accordion("✂️ Strategy Document Chunking", open=True): chunk_toggle = gr.Checkbox( value=True, label="Enable semantic chunking for strategy/document data" ) size_input = gr.Number( value=700, minimum=100, maximum=4096, step=50, label="Chunk size" ) overlap_input = gr.Number( value=100, minimum=0, maximum=1024, step=25, label="Chunk overlap" ) image_prompt = gr.Textbox( value=( "Describe only observable chart information. Separate " "observation from inference. Do not invent exact prices." ), lines=4, label="🖼️ Chart/Image Analysis Objective" ) decision_prompt = gr.Textbox( value=( "Evaluate the supplied trade proposal against the " "provided strategy rules and return an approval or " "denial decision only when an explicit label exists." ), lines=4, label="⚖️ Trade Approval / Denial Objective" ) run_btn = gr.Button( "🚀 Build Trading LLM Dataset", variant="primary" ) with gr.Column(scale=1): log_monitor = gr.Textbox( label="🖥️ Dataset Build Log", lines=14 ) rag_monitor = gr.Textbox( label="🔒 Private RAG Rules Retrieved", lines=8 ) download_btn = gr.DownloadButton( "💾 Download JSONL Dataset", variant="primary" ) run_btn.click( fn=execute_trading_dataset_pipeline, inputs=[ file_uploader, dataset_type, chunk_toggle, size_input, overlap_input, symbol_input, timeframe_input, image_prompt, decision_prompt, ], outputs=[log_monitor, download_btn, rag_monitor] ) with gr.Tab("🔒 Backend RAG Administration"): gr.Markdown( """ ### Private system knowledge base This panel is intended for the Space owner/administrator. Upload documents that define your dataset-generation standards, strategy-library schema, timestamp policy, numerical feature/label rules, image-to-text rules, and Unsloth training formats. **Important:** a production deployment should protect this tab with authentication. The RAG documents are stored under `/data/system_knowledge_base` and are never included in the user dataset download. """ ) admin_files = gr.File( file_count="multiple", type="filepath", label="📚 Upload private dataset-generation rule documents" ) rebuild_btn = gr.Button( "🔄 Import Documents & Rebuild Private RAG Index", variant="primary" ) admin_status = gr.Textbox( label="Backend RAG Status", lines=8 ) rebuild_btn.click( fn=rebuild_backend_rag_index, inputs=[admin_files], outputs=[admin_status] ) if __name__ == "__main__": build_private_rag_index() demo.launch()