Stage 04.1 - storage foundation with postgres

app/.env.example

@@ -4,6 +4,12 @@ APP_ENV=dev
 LOG_LEVEL=INFO
 TZ=Europe/Minsk
 
+DB_HOST=localhost
+DB_PORT=5432
+DB_NAME=dzentra_bot
+DB_USER=dzentra_bot
+DB_PASSWORD=change_me
+
 EXCHANGE_ENABLED=true
 EXCHANGE_NAME=dzengi
 EXCHANGE_BASE_URL=https://demo-api-adapter.dzengi.com

app/requirements.txt

@@ -1,2 +1,3 @@
 aiogram==3.13.1
 python-dotenv==1.0.1
+psycopg[binary]==3.2.9

app/src/bootstrap/app_factory.py

@@ -5,12 +5,14 @@ from aiogram.client.default import DefaultBotProperties
 
 from src.bootstrap.logging import setup_logging
 from src.core.config import load_settings
+from src.storage.schema import init_schema
 from src.telegram.routers import setup_routers
 
 
 def create_app() -> tuple[Bot, Dispatcher]:
     settings = load_settings()
     setup_logging(settings.log_level)
+    init_schema()
 
     bot = Bot(
         token=settings.bot_token,

app/src/core/config.py

@@ -21,6 +21,11 @@ class Settings:
     exchange_timeout_sec: int
     exchange_testnet: bool
     default_symbol: str
+    db_host: str
+    db_port: int
+    db_name: str
+    db_user: str
+    db_password: str
 def _parse_bool(raw_value: str, default: bool = False) -> bool:
     value = (raw_value or "").strip().lower()
     if not value:
@@ -49,4 +54,9 @@ def load_settings() -> Settings:
         exchange_timeout_sec=_parse_int(os.getenv("EXCHANGE_TIMEOUT_SEC", "10"), 10),
         exchange_testnet=_parse_bool(os.getenv("EXCHANGE_TESTNET", "false")),
         default_symbol=os.getenv("DEFAULT_SYMBOL", "BTC/USD_LEVERAGE").strip() or "BTC/USD_LEVERAGE",
+        db_host=os.getenv("DB_HOST", "localhost").strip() or "localhost",
+        db_port=_parse_int(os.getenv("DB_PORT", "5432"), 5432),
+        db_name=os.getenv("DB_NAME", "dzentra_bot").strip() or "dzentra_bot",
+        db_user=os.getenv("DB_USER", "dzentra_bot").strip() or "dzentra_bot",
+        db_password=os.getenv("DB_PASSWORD", "").strip(),
     )

app/src/core/system_status.py

@@ -6,6 +6,7 @@ from dataclasses import dataclass
 from src.core.config import load_settings
 from src.core.constants import APP_NAME, APP_VERSION
 from src.integrations.exchange.service import ExchangeService
+from src.storage.session import check_database_health
 
 
 @dataclass(slots=True)
@@ -45,6 +46,7 @@ def get_system_snapshot() -> SystemSnapshot:
 
     exchange_health = exchange_service.get_health()
     private_auth_health = exchange_service.get_private_auth_health()
+    db_ok, db_message = check_database_health()
 
     if exchange_health.ok and exchange_health.mode == "mock":
         exchange_state = "🟡 mock mode"
@@ -55,6 +57,7 @@ def get_system_snapshot() -> SystemSnapshot:
 
     symbol_state = "🟢 OK" if symbol_validation and symbol_validation.is_valid else "🔴 ошибка"
     private_auth_state = "🟢 OK" if private_auth_health.ok else "🔴 ошибка"
+    db_state = "🟢 OK" if db_ok else "🔴 ошибка"
 
     components = [
         ComponentStatus(
@@ -84,8 +87,8 @@ def get_system_snapshot() -> SystemSnapshot:
         ),
         ComponentStatus(
             name="База данных",
-            state="🟡 не подключена",
-            details="Слой хранения пока только подготовлен структурно.",
+            state=db_state,
+            details=db_message,
         ),
     ]
 

app/src/storage/models.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(slots=True)
+class BalanceSnapshotRecord:
+    id: int | None
+    created_at: str
+    source: str
+    payload_json: str
+
+
+@dataclass(slots=True)
+class JournalEventRecord:
+    id: int | None
+    created_at: str
+    level: str
+    event_type: str
+    message: str
+    payload_json: str | None
+
+
+@dataclass(slots=True)
+class OrderDraftRecord:
+    id: int | None
+    created_at: str
+    symbol: str
+    side: str
+    order_type: str
+    quantity: str
+    status: str
+    payload_json: str | None

app/src/storage/schema.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from src.storage.session import get_connection
+
+
+DDL = [
+    '''
+    CREATE TABLE IF NOT EXISTS balance_snapshots (
+        id BIGSERIAL PRIMARY KEY,
+        created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+        source TEXT NOT NULL,
+        payload_json JSONB NOT NULL
+    )
+    ''',
+    '''
+    CREATE TABLE IF NOT EXISTS journal_events (
+        id BIGSERIAL PRIMARY KEY,
+        created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+        level TEXT NOT NULL,
+        event_type TEXT NOT NULL,
+        message TEXT NOT NULL,
+        payload_json JSONB
+    )
+    ''',
+    '''
+    CREATE TABLE IF NOT EXISTS order_drafts (
+        id BIGSERIAL PRIMARY KEY,
+        created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+        symbol TEXT NOT NULL,
+        side TEXT NOT NULL,
+        order_type TEXT NOT NULL,
+        quantity NUMERIC(36, 18) NOT NULL,
+        status TEXT NOT NULL,
+        payload_json JSONB
+    )
+    '''
+]
+
+
+def init_schema() -> None:
+    with get_connection() as connection:
+        with connection.cursor() as cursor:
+            for statement in DDL:
+                cursor.execute(statement)

app/src/storage/session.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+
+import psycopg
+
+from src.core.config import load_settings
+
+
+def build_dsn() -> str:
+    settings = load_settings()
+    password_part = settings.db_password.replace("@", "%40")
+    return (
+        f"postgresql://{settings.db_user}:{password_part}"
+        f"@{settings.db_host}:{settings.db_port}/{settings.db_name}"
+    )
+
+
+@contextmanager
+def get_connection():
+    connection = psycopg.connect(build_dsn(), autocommit=False)
+    try:
+        yield connection
+        connection.commit()
+    finally:
+        connection.close()
+
+
+def check_database_health() -> tuple[bool, str]:
+    try:
+        with get_connection() as connection:
+            with connection.cursor() as cursor:
+                cursor.execute("SELECT current_database(), current_user, version()")
+                row = cursor.fetchone()
+                if row is None:
+                    return False, "PostgreSQL ping returned no rows."
+                db_name, db_user, version = row
+    except Exception as exc:
+        return False, f"PostgreSQL error: {exc}"
+
+    version_short = str(version).split(",")[0]
+    return True, f"PostgreSQL OK: db={db_name}, user={db_user}, {version_short}"