"""MCP tools for structured queries."""
from ..database.sqlite_client import SQLiteClient
from typing import Optional
class QueryTools:
"""Tools for structured database queries."""
def __init__(self, db_client: SQLiteClient):
"""Initialize query tools.
Args:
db_client: SQLite database client
"""
self.db = db_client
async def get_holdings_by_symbol(self, symbol: str) -> list[dict]:
"""Get all holdings for a specific symbol across all statements.
Args:
symbol: Stock/ETF symbol
Returns:
List of holdings
"""
return await self.db.get_holdings_by_symbol(symbol)
async def get_transactions_by_date_range(
self, start_date: str, end_date: str, account_number: Optional[str] = None
) -> list[dict]:
"""Get transactions within a date range.
Args:
start_date: Start date (YYYY-MM-DD)
end_date: End date (YYYY-MM-DD)
account_number: Optional account number filter
Returns:
List of transactions
"""
return await self.db.get_transactions_by_date_range(
start_date=start_date,
end_date=end_date,
account_number=account_number,
)
async def get_account_balance(self, account_number: str, date: str) -> Optional[dict]:
"""Get account balance at a specific date.
Args:
account_number: Account number
date: Date (YYYY-MM-DD)
Returns:
Statement data or None
"""
return await self.db.get_account_balance(account_number=account_number, date=date)
async def get_performance_history(
self, account_number: str, limit: int = 12
) -> dict:
"""Get performance history for an account over time.
Args:
account_number: Account number
limit: Number of months to return (default 12)
Returns:
Performance history with returns and balances
"""
history = await self.db.get_performance_history(account_number, limit)
if not history:
return {"error": f"No performance data found for account {account_number}"}
# Calculate some additional metrics from the history
balances = [h["current_balance_cad"] for h in history if h["current_balance_cad"]]
result = {
"account_number": account_number,
"periods": len(history),
"latest_balance": balances[0] if balances else None,
"history": history,
}
# Add simple statistics if we have enough data
if len(balances) >= 2:
result["balance_change"] = balances[0] - balances[-1]
result["balance_change_pct"] = (
(balances[0] - balances[-1]) / balances[-1] * 100
if balances[-1] != 0 else None
)
return result
async def compare_to_benchmark(
self, account_number: str, benchmark_name: Optional[str] = None
) -> dict:
"""Compare account performance to benchmarks.
Args:
account_number: Account number
benchmark_name: Optional specific benchmark to compare to
Returns:
Performance comparison data
"""
# Get latest statement for the account
history = await self.db.get_performance_history(account_number, limit=1)
if not history:
return {"error": f"No data found for account {account_number}"}
statement_id = history[0]["statement_id"]
account_returns = history[0]
# Get benchmarks for this statement
benchmarks = await self.db.get_benchmarks_for_statement(statement_id)
if not benchmarks:
return {
"account_number": account_number,
"account_returns": account_returns,
"benchmarks": [],
"message": "No benchmark data available for this statement"
}
# Filter by benchmark name if specified
if benchmark_name:
benchmarks = [b for b in benchmarks if benchmark_name.lower() in b["name"].lower()]
# Build comparison
comparisons = []
for benchmark in benchmarks:
comparison = {
"benchmark_name": benchmark["name"],
"benchmark_symbol": benchmark["symbol"],
}
# Add returns for each period
for period in ["ytd", "1y", "3y", "5y"]:
account_key = f"return_{period}"
benchmark_key = f"return_{period}"
alpha_key = f"alpha_{period}"
account_return = account_returns.get(account_key)
benchmark_return = benchmark.get(benchmark_key)
comparison[f"account_{period}"] = account_return
comparison[f"benchmark_{period}"] = benchmark_return
# Use stored alpha or calculate
if benchmark.get(alpha_key):
comparison[f"alpha_{period}"] = benchmark[alpha_key]
elif account_return is not None and benchmark_return is not None:
comparison[f"alpha_{period}"] = account_return - benchmark_return
comparisons.append(comparison)
return {
"account_number": account_number,
"statement_date": history[0].get("statement_date"),
"comparisons": comparisons
}
async def get_risk_metrics(self, account_number: str) -> dict:
"""Get risk metrics for an account.
Args:
account_number: Account number
Returns:
Risk metrics (volatility, beta, Sharpe ratio, max drawdown)
"""
metrics = await self.db.get_risk_metrics(account_number, limit=1)
if not metrics:
return {
"account_number": account_number,
"message": "No risk metrics calculated yet. Risk metrics require multiple months of data.",
"metrics": None
}
return {
"account_number": account_number,
"calculation_date": metrics[0]["calculation_date"],
"period_months": metrics[0]["period_months"],
"metrics": {
"volatility": metrics[0]["volatility"],
"beta": metrics[0]["beta"],
"sharpe_ratio": metrics[0]["sharpe_ratio"],
"max_drawdown": metrics[0]["max_drawdown"],
"var_95": metrics[0]["var_95"],
}
}
async def get_portfolio_summary(self) -> dict:
"""Get consolidated portfolio summary across all accounts.
Returns:
Summary with total value, allocation, and performance
"""
accounts = await self.db.get_accounts()
if not accounts:
return {"error": "No accounts found"}
# Calculate totals
total_balance = sum(
a["current_balance_cad"] for a in accounts
if a["current_balance_cad"]
)
# Build summary
account_summaries = []
for account in accounts:
balance = account["current_balance_cad"] or 0
summary = {
"account_number": account["account_number"],
"institution": account["institution"],
"latest_date": account["latest_date"],
"balance_cad": balance,
"allocation_pct": (balance / total_balance * 100) if total_balance > 0 else 0,
"return_ytd": account["return_ytd"],
}
account_summaries.append(summary)
# Calculate weighted average return
weighted_return = 0
total_with_return = 0
for account in accounts:
if account["current_balance_cad"] and account["return_ytd"]:
weighted_return += account["current_balance_cad"] * account["return_ytd"]
total_with_return += account["current_balance_cad"]
avg_return_ytd = weighted_return / total_with_return if total_with_return > 0 else None
return {
"total_accounts": len(accounts),
"total_balance_cad": total_balance,
"weighted_avg_return_ytd": avg_return_ytd,
"accounts": account_summaries
}
