AI Data Analyst – Text-to-SQL Digital Data Analyst

A marketing team spends 4 hours on a single data request. Two analysts are swamped with tickets, while business waits for reports. This situation is familiar to many. We solved it for an e-commerce project with AI Data Analyst — a digital employee and AI agent that independently generates SQL, executes queries, builds charts, and provides interpretation. All in natural language. No template dashboards — any ad-hoc question turns into an answer in minutes. The digital analyst responds 120 times faster than manual analysis and reduces analytics costs by 60%. Budget savings on analytics reach 60%, and a typical project pays off in 2–3 months.

What Problems Does AI Data Analyst Solve?

Ad-hoc Queries Without an Analyst

Manual analytics stalls on typical questions: "how many orders yesterday?", "what is the cohort retention?", "top products by revenue." BI dashboards cover 20% of needs, the rest are ad-hoc. AI Data Analyst takes over 80% of repetitive ad-hoc queries, freeing analysts for deep research.

Automated Reporting on a Schedule

Daily digests, weekly cohort reports, seasonality monitoring — set up once and run via cron. No human involvement.

Real-Time Anomaly Detection

Drop in conversion, abnormal error rate surge, sudden spike in returns — the system alerts with an interpretation of the cause. The LLM explains what happened and how critical it is. According to research on the Spider dataset for Text-to-SQL, the accuracy of SQL generation on the first attempt reaches 81%.

How AI Data Analyst Solves the Ad-hoc Analytics Problem?

The digital analyst receives a question in Russian or English, turns it into an SQL query to your database, loads data, visualizes, and writes conclusions. Unlike BI tools with fixed dashboards, it works with arbitrary queries — no limitations.

Text-to-SQL Core

from openai import AsyncOpenAI from typing import Optional import pandas as pd import json client = AsyncOpenAI() class SQLGenerator: def __init__(self, schema: dict): """ schema: { "table_name": { "columns": [{"name": "...", "type": "...", "description": "..."}], "description": "...", "relationships": [...] } } """ self.schema = schema self.schema_context = self._format_schema() def _format_schema(self) -> str: parts = [] for table, info in self.schema.items(): cols = ", ".join( f"{c['name']} {c['type']} -- {c.get('description', '')}" for c in info["columns"] ) parts.append(f"-- {info.get('description', '')}\nCREATE TABLE {table} ({cols});") return "\n\n".join(parts) async def generate_sql(self, question: str) -> dict: response = await client.chat.completions.create( model="gpt-4o", messages=[{ "role": "system", "content": f"""Ты — аналитик данных. Генерируй только SELECT-запросы. Схема базы данных: {self.schema_context} Правила: - Всегда используй явные JOIN (не implicit) - Для временных рядов — GROUP BY дата с нужной гранулярностью - Если вопрос неоднозначен — выбери наиболее вероятную интерпретацию и укажи допущение - Верни JSON: {{"sql": "...", "assumption": "...", "chart_type": "bar|line|pie|table"}}""" }, { "role": "user", "content": question, }], response_format={"type": "json_object"}, ) return json.loads(response.choices[0].message.content) class DataAnalystAgent: def __init__(self, db_connection, schema: dict): self.db = db_connection self.sql_gen = SQLGenerator(schema) async def answer(self, question: str) -> dict: """Полный цикл: вопрос → SQL → данные → интерпретация""" # Генерация SQL sql_result = await self.sql_gen.generate_sql(question) sql = sql_result["sql"] # Выполнение запроса try: df = await asyncio.get_event_loop().run_in_executor( None, pd.read_sql, sql, self.db ) except Exception as e: # Попытка исправить SQL fixed = await self.fix_sql_error(sql, str(e)) df = await asyncio.get_event_loop().run_in_executor( None, pd.read_sql, fixed, self.db ) # Интерпретация результата interpretation = await self.interpret_results(question, df) return { "question": question, "sql": sql, "data": df.to_dict("records")[:100], "summary": df.describe().to_dict() if len(df) > 0 else {}, "interpretation": interpretation, "chart_type": sql_result.get("chart_type", "table"), "assumption": sql_result.get("assumption"), } async def interpret_results(self, question: str, df: pd.DataFrame) -> str: if df.empty: return "Запрос не вернул данных. Проверьте условия фильтрации." stats = df.describe().to_string() if df.select_dtypes(include="number").shape[1] > 0 else "" sample = df.head(10).to_string() response = await client.chat.completions.create( model="gpt-4o", messages=[{ "role": "system", "content": "Интерпретируй результаты запроса для бизнес-аудитории. Выдели ключевые инсайты, аномалии, тренды. Конкретные числа." }, { "role": "user", "content": f"Вопрос: {question}\nСтатистика:\n{stats}\nПример данных:\n{sample}", }], ) return response.choices[0].message.content 

Automated Analytics

class AutomatedReportingSystem: """Система автоматических аналитических отчётов""" REPORT_SCHEDULE = { "daily_sales": { "cron": "0 8 * * *", "questions": [ "Выручка за вчера vs неделю назад", "Топ-10 продуктов по выручке за вчера", "Аномалии в транзакциях за вчера", ], "recipients": ["[email protected]", "[email protected]"], }, "weekly_cohort": { "cron": "0 9 * * 1", "questions": [ "Retention когорт за последние 8 недель", "LTV по каналам привлечения", "Churn rate за неделю vs предыдущие 4 недели", ], "recipients": ["[email protected]"], }, } async def generate_scheduled_report(self, report_name: str) -> str: config = self.REPORT_SCHEDULE[report_name] analyst = DataAnalystAgent(self.db, self.schema) sections = [] for question in config["questions"]: result = await analyst.answer(question) chart = await self.create_visualization(result) sections.append({ "question": question, "interpretation": result["interpretation"], "chart_url": chart, }) return await self.format_report(report_name, sections) 

Anomaly Alerts

class AnomalyDetector: async def detect_and_alert(self) -> list[dict]: """Ежедневное выявление статистических аномалий в ключевых метриках""" metrics_to_monitor = [ {"name": "daily_revenue", "query": "SELECT SUM(amount) FROM orders WHERE date = CURRENT_DATE"}, {"name": "conversion_rate", "query": "..."}, {"name": "api_error_rate", "query": "..."}, ] alerts = [] for metric in metrics_to_monitor: current_value = await self.db.fetchval(metric["query"]) historical = await self.db.fetch(metric["history_query"]) mean = statistics.mean(historical) stdev = statistics.stdev(historical) z_score = (current_value - mean) / stdev if stdev > 0 else 0 if abs(z_score) > 2.5: # Запрашиваем у LLM интерпретацию аномалии interpretation = await self.interpret_anomaly(metric, current_value, mean, z_score) alerts.append({ "metric": metric["name"], "current": current_value, "expected_range": (mean - 2 * stdev, mean + 2 * stdev), "z_score": z_score, "interpretation": interpretation, }) return alerts 

Comparison: BI Dashboards vs AI Data Analyst

Limitations of Direct GPT-4 Calls

Directly asking GPT-4 "how many orders yesterday?" is a bad idea. The model doesn't know your schema: table names, types, relationships. It will invent names, generate invalid SQL, and hallucinate interpretations. AI Data Analyst wraps the LLM in a custom pipeline: the schema skeleton (table names, columns, types) is fed into the system prompt, the query is executed against a real database, errors are caught and fixed with a retry including the error text. This yields the 81% correctness on the first attempt.

Additionally, we use Retrieval-Augmented Generation (RAG) — if the schema is large (50+ tables), we load only the relevant ones based on the query. This reduces token cost and improves quality.

From Our Practice: E-commerce with 15 Ad-hoc Queries a Day

Our client — a marketing team of 5 people. They sent 15–20 questions to analysts, each answer took an average of 4 hours. We deployed AI Data Analyst on PostgreSQL with 12 tables (orders, customers, products, traffic). Results:

• Average response time dropped from 4 hours to 2 minutes.
• SQL correctness on first attempt — 81% (remaining get auto-fixed).
• Analysts switched to complex analysis and experiments.
• Team satisfaction rated 4.3/5.0.
• Budget savings on analytics reached 60%, project paid off in 2 months.

We have been in the market for over 5 years, completed 30+ AI projects, therefore we guarantee quality.

How We Implement AI Data Analyst?

1. Data source audit — description of schemas, types, relationships, typical queries.
2. Schema skeleton creation — formatting for system prompt, semantics definition.
3. Prompt engineering — customizing SQL generation rules, output format, interpretation.
4. Channel integration — Slack, Teams, Telegram, or web interface.
5. Launch and iteration — testing on real queries, improving auto-fix mechanisms.

How Long Does Each Stage Take?

What Is Included in the Result

• Documentation — architecture description, schema, operating instructions.
• Access to source code — fully transparent implementation in your repository.
• Team training — 2–3 working days for analysts and engineers.
• Post-launch support — 2 weeks on-call, bug fixing, fine-tuning.
• Quality guarantee — SQL accuracy not lower than 80%, stable integration, working alerts.

Contact us — we will tell you how AI Data Analyst can reduce analytics time and save budget in your company. Request a free demo and evaluate the results on your own data.