"""Example: Index a statement and query it."""
import asyncio
import sys
import os
# Add parent directory to path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.database.sqlite_client import SQLiteClient
from src.database.lancedb_client import LanceDBClient
from src.embeddings.generator import EmbeddingGenerator
from src.tools.indexing import IndexingTools
from src.tools.search import SearchTools
from src.tools.query import QueryTools
async def main():
"""Main example function."""
# Initialize clients (paths auto-resolved relative to module location)
print("Initializing clients...")
sqlite_client = SQLiteClient()
lance_client = LanceDBClient()
embedding_gen = EmbeddingGenerator(model_name="all-MiniLM-L6-v2")
# Initialize database
await sqlite_client.initialize()
# Initialize tools
indexing_tools = IndexingTools(
sqlite_client=sqlite_client,
lance_client=lance_client,
embedding_gen=embedding_gen,
)
search_tools = SearchTools(lance_client=lance_client, embedding_gen=embedding_gen)
query_tools = QueryTools(db_client=sqlite_client)
# Path to sample PDF
pdf_path = "/Users/jasonlo/Nextcloud/00_Personal/20_Finances/20_Financial_Statements/Investment Statements/Questrade/Questrade TFSA-51516162 Statement 2025-09.pdf"
if not os.path.exists(pdf_path):
print(f"Error: PDF not found at {pdf_path}")
return
# Index the statement
print(f"\nIndexing statement: {pdf_path}")
result = await indexing_tools.index_statement(pdf_path)
if result["status"] == "success":
print(f"✓ Successfully indexed statement!")
print(f" Statement ID: {result['statement_id']}")
print(f" Institution: {result['institution']}")
print(f" Account: {result['account_number']}")
print(f" Date: {result['statement_date']}")
else:
print(f"✗ Error: {result.get('message')}")
return
# Get indexing stats
print("\nIndexing Statistics:")
stats = await indexing_tools.get_indexing_stats()
print(f" Total Statements: {stats.get('total_statements', 0)}")
print(f" Total Holdings: {stats.get('total_holdings', 0)}")
print(f" Total Transactions: {stats.get('total_transactions', 0)}")
print(f" Total Chunks: {stats.get('total_chunks', 0)}")
print(f" Unique Symbols: {stats.get('unique_symbols', 0)}")
# Search with natural language
print("\n" + "="*80)
print("Natural Language Search Examples:")
print("="*80)
queries = [
"What is the SPY holding value?",
"Show me cash balances",
"What were the dividends?",
]
for query in queries:
print(f"\nQuery: '{query}'")
results = await search_tools.search_statements(query=query, limit=3)
if results:
for i, result in enumerate(results, 1):
print(f"\n Result {i}:")
print(f" Type: {result['chunk_type']}")
print(f" Account: {result['account_number']}")
print(f" Date: {result['statement_date']}")
print(f" Content: {result['content'][:200]}...")
else:
print(" No results found.")
# Structured queries
print("\n" + "="*80)
print("Structured Query Examples:")
print("="*80)
# Get holdings by symbol
print("\nQuery: Get holdings for symbol 'SPY'")
holdings = await query_tools.get_holdings_by_symbol("SPY")
if holdings:
for holding in holdings:
print(f" Date: {holding['statement_date']}")
print(f" Quantity: {holding['quantity']}")
print(f" Market Value: ${holding['market_value_cad']:,.2f} CAD")
print(f" P/L: ${holding['profit_loss']:,.2f} ({holding['percent_return']:.2f}%)")
else:
print(" No holdings found.")
# Get transactions by date range
print("\nQuery: Get transactions from 2025-01-01 to 2025-12-31")
transactions = await query_tools.get_transactions_by_date_range(
start_date="2025-01-01",
end_date="2025-12-31"
)
if transactions:
print(f" Found {len(transactions)} transactions")
for txn in transactions[:5]: # Show first 5
print(f" {txn['transaction_date']}: {txn['activity_type']} {txn['description']} ${txn['net_amount']:,.2f}")
else:
print(" No transactions found.")
# Get account balance
print("\nQuery: Get account balance as of 2025-09-30")
balance = await query_tools.get_account_balance("51516162", "2025-09-30")
if balance:
print(f" Account: {balance['account_number']}")
print(f" Statement Date: {balance['statement_date']}")
print(f" Institution: {balance['institution']}")
else:
print(" No balance information found.")
print("\n" + "="*80)
print("Example complete!")
print("="*80)
if __name__ == "__main__":
asyncio.run(main())
