-- Accounts table
CREATE TABLE accounts (
    account_id SERIAL PRIMARY KEY,
    customer_id INTEGER NOT NULL,
    account_type VARCHAR(20) CHECK (account_type IN ('CHECKING', 'SAVINGS', 'TRADING')),
    currency VARCHAR(3) DEFAULT 'USD',
    balance DECIMAL(18,2) DEFAULT 0.00,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    status VARCHAR(20) DEFAULT 'ACTIVE',
    FOREIGN KEY (customer_id) REFERENCES customers(customer_id)
);

-- Transactions table
CREATE TABLE transactions (
    transaction_id SERIAL PRIMARY KEY,
    account_id INTEGER NOT NULL,
    transaction_type VARCHAR(20) CHECK (transaction_type IN ('DEPOSIT', 'WITHDRAWAL', 'TRANSFER', 'TRADE')),
    amount DECIMAL(18,2) NOT NULL,
    currency VARCHAR(3) DEFAULT 'USD',
    description TEXT,
    transaction_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    reference_id VARCHAR(50),
    status VARCHAR(20) DEFAULT 'PENDING',
    FOREIGN KEY (account_id) REFERENCES accounts(account_id)
);

-- Orders table (for trading)
CREATE TABLE orders (
    order_id SERIAL PRIMARY KEY,
    account_id INTEGER NOT NULL,
    symbol VARCHAR(20) NOT NULL,
    side VARCHAR(4) CHECK (side IN ('BUY', 'SELL')),
    quantity INTEGER NOT NULL,
    price DECIMAL(18,2),
    order_type VARCHAR(20) CHECK (order_type IN ('MARKET', 'LIMIT', 'STOP')),
    status VARCHAR(20) DEFAULT 'PENDING',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    filled_quantity INTEGER DEFAULT 0,
    filled_price DECIMAL(18,2),
    FOREIGN KEY (account_id) REFERENCES accounts(account_id)
);

-- Price history table
CREATE TABLE price_history (
    price_id SERIAL PRIMARY KEY,
    symbol VARCHAR(20) NOT NULL,
    price DECIMAL(18,6) NOT NULL,
    volume INTEGER,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE(symbol, timestamp)
);

-- Create indexes for performance
CREATE INDEX idx_transactions_account_id ON transactions(account_id);
CREATE INDEX idx_transactions_date ON transactions(transaction_date);
CREATE INDEX idx_orders_account_id ON orders(account_id);
CREATE INDEX idx_orders_symbol ON orders(symbol);
CREATE INDEX idx_orders_status ON orders(status);
CREATE INDEX idx_price_history_symbol ON price_history(symbol);
CREATE INDEX idx_price_history_timestamp ON price_history(timestamp);

-- Get account balance with transaction history
SELECT 
    a.account_id,
    a.balance,
    COUNT(t.transaction_id) as transaction_count,
    SUM(t.amount) as total_volume
FROM accounts a
LEFT JOIN transactions t ON a.account_id = t.account_id
WHERE a.account_id = 123
GROUP BY a.account_id, a.balance;

-- Calculate daily P&L for a trading account
SELECT 
    DATE(transaction_date) as trade_date,
    SUM(CASE 
        WHEN transaction_type = 'TRADE' AND amount > 0 THEN amount 
        ELSE 0 
    END) as total_buys,
    SUM(CASE 
        WHEN transaction_type = 'TRADE' AND amount < 0 THEN amount 
        ELSE 0 
    END) as total_sells,
    SUM(amount) as net_pnl
FROM transactions
WHERE account_id = 123 
    AND transaction_type = 'TRADE'
GROUP BY DATE(transaction_date)
ORDER BY trade_date DESC;

-- Find the most active traders
SELECT 
    c.customer_id,
    c.name,
    COUNT(t.transaction_id) as trade_count,
    SUM(ABS(t.amount)) as total_volume
FROM customers c
JOIN accounts a ON c.customer_id = a.customer_id
JOIN transactions t ON a.account_id = t.account_id
WHERE t.transaction_type = 'TRADE'
GROUP BY c.customer_id, c.name
ORDER BY trade_count DESC
LIMIT 10;

-- Calculate moving average using window functions
SELECT 
    symbol,
    timestamp,
    price,
    AVG(price) OVER (PARTITION BY symbol ORDER BY timestamp ROWS BETWEEN 49 PRECEDING AND CURRENT ROW) as ma_50,
    AVG(price) OVER (PARTITION BY symbol ORDER BY timestamp ROWS BETWEEN 199 PRECEDING AND CURRENT ROW) as ma_200
FROM price_history
WHERE symbol = 'AAPL'
ORDER BY timestamp DESC;

-- Detect unusual trading activity (3 standard deviations from mean)
WITH daily_stats AS (
    SELECT 
        account_id,
        DATE(transaction_date) as trade_date,
        SUM(amount) as daily_volume,
        AVG(SUM(amount)) OVER (PARTITION BY account_id) as avg_volume,
        STDDEV(SUM(amount)) OVER (PARTITION BY account_id) as std_volume
    FROM transactions
    WHERE transaction_type = 'TRADE'
    GROUP BY account_id, DATE(transaction_date)
)
SELECT *
FROM daily_stats
WHERE ABS(daily_volume - avg_volume) > 3 * std_volume;

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey, DECIMAL
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from datetime import datetime

Base = declarative_base()

class Account(Base):
    __tablename__ = 'accounts'
    
    account_id = Column(Integer, primary_key=True)
    customer_id = Column(Integer, ForeignKey('customers.customer_id'))
    account_type = Column(String(20))
    currency = Column(String(3), default='USD')
    balance = Column(DECIMAL(18,2), default=0.00)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    status = Column(String(20), default='ACTIVE')
    
    transactions = relationship('Transaction', back_populates='account')
    orders = relationship('Order', back_populates='account')

class Transaction(Base):
    __tablename__ = 'transactions'
    
    transaction_id = Column(Integer, primary_key=True)
    account_id = Column(Integer, ForeignKey('accounts.account_id'))
    transaction_type = Column(String(20))
    amount = Column(DECIMAL(18,2))
    currency = Column(String(3), default='USD')
    description = Column(String(255))
    transaction_date = Column(DateTime, default=datetime.utcnow)
    reference_id = Column(String(50))
    status = Column(String(20), default='PENDING')
    
    account = relationship('Account', back_populates='transactions')

class Order(Base):
    __tablename__ = 'orders'
    
    order_id = Column(Integer, primary_key=True)
    account_id = Column(Integer, ForeignKey('accounts.account_id'))
    symbol = Column(String(20))
    side = Column(String(4))
    quantity = Column(Integer)
    price = Column(DECIMAL(18,2))
    order_type = Column(String(20))
    status = Column(String(20), default='PENDING')
    created_at = Column(DateTime, default=datetime.utcnow)
    filled_quantity = Column(Integer, default=0)
    filled_price = Column(DECIMAL(18,2))
    
    account = relationship('Account', back_populates='orders')

# Database connection
engine = create_engine('postgresql://user:password@localhost/fintech')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

# Usage
def create_transaction(account_id, transaction_type, amount, description):
    session = Session()
    try:
        transaction = Transaction(
            account_id=account_id,
            transaction_type=transaction_type,
            amount=amount,
            description=description
        )
        session.add(transaction)
        
        # Update account balance
        account = session.query(Account).filter_by(account_id=account_id).first()
        if account:
            account.balance += amount
            account.updated_at = datetime.utcnow()
        
        session.commit()
        return transaction.transaction_id
    except Exception as e:
        session.rollback()
        raise e
    finally:
        session.close()

from pymongo import MongoClient
from datetime import datetime

client = MongoClient('mongodb://localhost:27017/')
db = client['fintech']

# Collection: trades
trades_collection = db['trades']

# Document structure
trade_document = {
    'trade_id': 'TRD-2024-001',
    'symbol': 'AAPL',
    'side': 'BUY',
    'quantity': 100,
    'price': 150.50,
    'timestamp': datetime.utcnow(),
    'account_id': 'ACC-123',
    'order_id': 'ORD-456',
    'execution_details': {
        'venue': 'NASDAQ',
        'order_type': 'LIMIT',
        'time_in_force': 'DAY'
    },
    'counterparty': 'CITADEL',
    'tags': ['tech', 'blue-chip', 'large-cap']
}

# Insert
result = trades_collection.insert_one(trade_document)

# Query
trades = trades_collection.find({
    'symbol': 'AAPL',
    'timestamp': {'$gte': datetime(2024, 1, 1)}
})

# Aggregation
pipeline = [
    {'$match': {'symbol': 'AAPL'}},
    {'$group': {
        '_id': '$symbol',
        'total_volume': {'$sum': '$quantity'},
        'average_price': {'$avg': '$price'},
        'max_price': {'$max': '$price'},
        'min_price': {'$min': '$price'}
    }}
]
aggregation_result = trades_collection.aggregate(pipeline)

# Portfolio document
portfolio_document = {
    'portfolio_id': 'PORT-001',
    'account_id': 'ACC-123',
    'name': 'Growth Portfolio',
    'created_at': datetime(2024, 1, 1),
    'holdings': [
        {
            'symbol': 'AAPL',
            'quantity': 1000,
            'average_cost': 145.00,
            'current_price': 150.50,
            'sector': 'Technology',
            'weight': 0.25
        },
        {
            'symbol': 'GOOGL',
            'quantity': 500,
            'average_cost': 2750.00,
            'current_price': 2800.00,
            'sector': 'Technology',
            'weight': 0.30
        }
    ],
    'performance': {
        'total_return': 0.08,
        'ytd_return': 0.12,
        'volatility': 0.18,
        'sharpe': 0.45
    },
    'risk_metrics': {
        'var_95': 0.05,
        'cvar_95': 0.08,
        'beta': 1.2,
        'max_drawdown': 0.15
    }
}

# Query holdings
portfolio = portfolios_collection.find_one({'account_id': 'ACC-123'})
for holding in portfolio['holdings']:
    print(f"{holding['symbol']}: {holding['quantity']} shares")

# Update price for all holdings
portfolios_collection.update_many(
    {},
    {'$set': {'holdings.$[].current_price': 152.00}}
)

# Complex query: find portfolios with >10% AAPL weight
portfolios = portfolios_collection.find({
    'holdings': {
        '$elemMatch': {
            'symbol': 'AAPL',
            'weight': {'$gt': 0.10}
        }
    }
})

from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import pandas as pd

# Connect
client = InfluxDBClient(
    url='http://localhost:8086',
    token='my-token',
    org='my-org'
)

write_api = client.write_api(write_options=SYNCHRONOUS)

# Write tick data
point = Point('stock_price') \
    .tag('symbol', 'AAPL') \
    .tag('exchange', 'NASDAQ') \
    .field('price', 150.50) \
    .field('volume', 1000000) \
    .time(datetime.utcnow())

write_api.write(bucket='fintech', record=point)

# Write multiple points
points = []
for i in range(100):
    price = 150 + np.random.randn() * 0.5
    point = Point('stock_price') \
        .tag('symbol', 'AAPL') \
        .field('price', price) \
        .time(datetime.utcnow() - pd.Timedelta(minutes=i))
    points.append(point)

write_api.write(bucket='fintech', record=points)

query_api = client.query_api()

# Query with Flux
query = '''
from(bucket: "fintech")
  |> range(start: -1d)
  |> filter(fn: (r) => r._measurement == "stock_price")
  |> filter(fn: (r) => r.symbol == "AAPL")
  |> aggregateWindow(every: 1m, fn: mean)
  |> yield(name: "mean")
'''

result = query_api.query(query)

# Convert to DataFrame
def influx_to_dataframe(result):
    data = []
    for table in result:
        for record in table.records:
            data.append({
                'time': record.get_time(),
                'symbol': record.values.get('symbol'),
                'price': record.get_value()
            })
    return pd.DataFrame(data)

df = influx_to_dataframe(result)

# Calculate statistics
query = '''
from(bucket: "fintech")
  |> range(start: -30d)
  |> filter(fn: (r) => r._measurement == "stock_price")
  |> filter(fn: (r) => r.symbol == "AAPL")
  |> aggregateWindow(every: 1h, fn: mean)
  |> derivative()
  |> histogram(bins: 50)
'''

# Real-time monitoring query
query = '''
from(bucket: "fintech")
  |> range(start: -1m)
  |> filter(fn: (r) => r._measurement == "stock_price")
  |> filter(fn: (r) => r.symbol == "AAPL")
  |> last()
  |> yield(name: "last_price")
'''

from influxdb_client.domain.retention_rule import RetentionRule

# Create retention policy
retention = RetentionRule(
    every_seconds=30 * 24 * 3600,  # 30 days
    shard_group_duration_seconds=24 * 3600  # 1 day
)

# Create bucket with retention
client.buckets_api().create_bucket(
    name='fintech_long_term',
    org='my-org',
    retention_rules=retention
)

# Continuous query for aggregation
create_continuous_query = '''
CREATE CONTINUOUS QUERY "cq_aggregates" ON "fintech"
BEGIN
  SELECT mean(price) as avg_price, max(price) as max_price, min(price) as min_price
  INTO "fintech_aggregated"."autogen"."price_1h"
  FROM "stock_price"
  GROUP BY time(1h), symbol
END
'''

-- B-tree index for range queries
CREATE INDEX idx_price_date ON price_history(symbol, timestamp);

-- Partial index for active orders
CREATE INDEX idx_active_orders ON orders(account_id) 
WHERE status = 'PENDING';

-- GiST index for spatial data (if applicable)
CREATE INDEX idx_location ON customers USING GIST(location);

-- BRIN index for large tables
CREATE INDEX idx_price_timestamp_brin ON price_history USING BRIN(timestamp);

-- Range partitioning by date
CREATE TABLE price_history_partitioned (
    price_id SERIAL,
    symbol VARCHAR(20),
    price DECIMAL(18,6),
    timestamp TIMESTAMP
) PARTITION BY RANGE (timestamp);

-- Create monthly partitions
CREATE TABLE price_history_2024_01 PARTITION OF price_history_partitioned
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
    
CREATE TABLE price_history_2024_02 PARTITION OF price_history_partitioned
    FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');

-- Hash partitioning for account data
CREATE TABLE accounts_partitioned (
    account_id SERIAL,
    customer_id INTEGER,
    balance DECIMAL(18,2)
) PARTITION BY HASH (account_id);

CREATE TABLE accounts_0 PARTITION OF accounts_partitioned
    FOR VALUES WITH (MODULUS 4, REMAINDER 0);

# Application-level sharding
class ShardingStrategy:
    def __init__(self, num_shards=8):
        self.num_shards = num_shards
    
    def get_shard(self, account_id):
        return account_id % self.num_shards
    
    def get_connection(self, account_id):
        shard = self.get_shard(account_id)
        return f"shard_{shard}"

# Usage
sharding = ShardingStrategy(num_shards=8)
account_id = 12345
shard = sharding.get_shard(account_id)
connection = sharding.get_connection(account_id)

from typing import Dict, List, Optional
import psycopg2
import pandas as pd
from datetime import datetime

class DatabaseManager:
    """
    Manages database connections and operations for a trading system.
    """
    
    def __init__(self, config: Dict):
        self.config = config
        self.connections = {}
    
    def get_connection(self, db_name: str):
        """Get or create a database connection."""
        if db_name not in self.connections:
            self.connections[db_name] = psycopg2.connect(
                host=self.config['host'],
                port=self.config['port'],
                database=db_name,
                user=self.config['user'],
                password=self.config['password']
            )
        return self.connections[db_name]
    
    def execute_query(self, db_name: str, query: str, params: Optional[tuple] = None):
        """Execute a query and return results."""
        conn = self.get_connection(db_name)
        cursor = conn.cursor()
        
        try:
            cursor.execute(query, params)
            if query.strip().upper().startswith('SELECT'):
                columns = [desc[0] for desc in cursor.description]
                data = cursor.fetchall()
                return pd.DataFrame(data, columns=columns)
            else:
                conn.commit()
                return cursor.rowcount
        finally:
            cursor.close()
    
    def batch_insert(self, db_name: str, table: str, data: pd.DataFrame):
        """Bulk insert data into a table."""
        conn = self.get_connection(db_name)
        cursor = conn.cursor()
        
        # Generate insert statement
        columns = ', '.join(data.columns)
        placeholders = ', '.join(['%s'] * len(data.columns))
        query = f"INSERT INTO {table} ({columns}) VALUES ({placeholders})"
        
        # Convert DataFrame to list of tuples
        values = [tuple(row) for row in data.to_numpy()]
        
        try:
            cursor.executemany(query, values)
            conn.commit()
            return len(values)
        finally:
            cursor.close()
    
    def close_all(self):
        """Close all connections."""
        for conn in self.connections.values():
            conn.close()
        self.connections.clear()

# Usage example
def setup_trading_database(config):
    """
    Set up the complete database schema for a trading system.
    """
    db_manager = DatabaseManager(config)
    
    # Create tables
    queries = [
        """
        CREATE TABLE IF NOT EXISTS customers (
            customer_id SERIAL PRIMARY KEY,
            email VARCHAR(255) UNIQUE NOT NULL,
            name VARCHAR(255),
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """,
        """
        CREATE TABLE IF NOT EXISTS accounts (
            account_id SERIAL PRIMARY KEY,
            customer_id INTEGER REFERENCES customers(customer_id),
            account_type VARCHAR(20),
            balance DECIMAL(18,2) DEFAULT 0.00,
            currency VARCHAR(3) DEFAULT 'USD',
            status VARCHAR(20) DEFAULT 'ACTIVE'
        )
        """,
        """
        CREATE TABLE IF NOT EXISTS orders (
            order_id SERIAL PRIMARY KEY,
            account_id INTEGER REFERENCES accounts(account_id),
            symbol VARCHAR(20),
            side VARCHAR(4),
            quantity INTEGER,
            price DECIMAL(18,2),
            order_type VARCHAR(20),
            status VARCHAR(20),
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            filled_quantity INTEGER DEFAULT 0,
            filled_price DECIMAL(18,2)
        )
        """,
        """
        CREATE TABLE IF NOT EXISTS executions (
            execution_id SERIAL PRIMARY KEY,
            order_id INTEGER REFERENCES orders(order_id),
            quantity INTEGER,
            price DECIMAL(18,2),
            execution_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            venue VARCHAR(20),
            execution_id_external VARCHAR(50)
        )
        """
    ]
    
    for query in queries:
        db_manager.execute_query('trading_db', query)
    
    return db_manager