### Configuration Settings for Trading Robot (Python)

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Defines the robot's behavior, trading parameters, and strategy settings in a Python configuration file. It includes Tinkoff API authentication details, instruments to trade, trading logic (order opening, reversal, lot count, profit targets), strategy parameters (profile and signal periods, volume level triggers), notification setup, and visualization options.

```python
# settings.py configuration example

# Tinkoff API authentication
TOKEN = "your_tinkoff_token_here"
IS_SANDBOX = True
ACCOUNT_ID = "your_account_id"

# Instruments to analyze and trade
INSTRUMENTS = [
    {"name": "USD000UTSTOM", "alias": "USD/RUB", "figi": "BBG0013HGFT4", "future": "BBG00VHGV1J0"},
    {"name": "SBER", "figi": "BBG004730N88", "future": "FUTSBRF06220"},
]

# Trading behavior
CAN_OPEN_ORDERS = True  # False = advisory mode, True = auto-trading
CAN_REVERSE_ORDER = True  # Close and reverse position on opposite signal
COUNT_LOTS = 2  # Number of lots per entry
COUNT_GOALS = 2  # Number of take-profit targets
FIRST_GOAL = 3  # First TP = stop_loss * 3
GOAL_STEP = 0.5  # Each subsequent goal increases by this factor
PERCENTAGE_STOP_LOSS = 0.03  # Stop-loss at 3% from max volume level

# Strategy parameters
PROFILE_PERIOD = "1h"  # Volume profile calculation period
SIGNAL_CLUSTER_PERIOD = "5min"  # Timeframe for signal candles
FIRST_TOUCH_VOLUME_LEVEL = 90  # Minutes before first touch is valid
SECOND_TOUCH_VOLUME_LEVEL = 5  # Minutes between subsequent touches
PERCENTAGE_VOLUME_LEVEL_RANGE = 0.03  # Price range around volume level (3%)

# Notification setup
NOTIFICATION = {
    "bot_token": "your_telegram_bot_token",
    "chat_id": "your_chat_id"
}

# Visualization
IS_SHOW_CHART = False  # Display interactive chart during backtesting

```

--------------------------------

### TradingRobot Main Class Initialization and Execution (Python)

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Initializes and runs the core TradingRobot class, which orchestrates data streaming, strategy execution, and order management for multiple instruments. It automatically sets up services, strategies, data persistence, and begins market data streaming. The robot syncs historical data, processes real-time trades, creates orders based on strategy signals, manages open positions, and saves statistics.

```python
from trading_robot import TradingRobot
import asyncio
from settings import INSTRUMENTS, TOKEN

# Initialize and run the trading robot
robot = TradingRobot()

# The robot automatically:
# - Creates OrderService with notification capabilities
# - Initializes ProfileTouchStrategy for each instrument
# - Sets up data persistence to CSV files
# - Starts streaming market data

asyncio.run(robot.main())

# Expected behavior:
# - Syncs historical trade data on startup
# - Streams real-time trades for configured instruments
# - Analyzes each trade through the strategy
# - Creates orders when entry conditions are met
# - Monitors open positions for stop-loss/take-profit
# - Saves statistics when trading day ends
```

--------------------------------

### Synchronize and Backfill Historical Trade Data

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Asynchronously loads and merges historical trade data with real-time streams for strategy initialization. It fetches trades hour by hour in reverse chronological order and combines them with an existing data file, ensuring a comprehensive dataset. Requires API token and specific libraries like pandas and tinkoff.invest.

```python
import asyncio
from tinkoff.invest import AsyncClient
from datetime import timedelta
from tinkoff.invest.utils import now
from constants import ONE_HOUR_TO_MINUTES
import pandas as pd

async def sync_and_backfill():
    TOKEN = "your_token"

    async with AsyncClient(TOKEN) as client:
        instrument = {
            "name": "SBER",
            "figi": "BBG004730N88"
        }

        # Load recent historical trades
        history_df = pd.DataFrame()
        current_date = now()
        time_offset = 0

        # Fetch trades hour by hour going backwards
        for _ in range(24):  # Get last 24 hours
            interval_from = current_date - timedelta(minutes=time_offset + ONE_HOUR_TO_MINUTES)
            interval_to = current_date - timedelta(minutes=time_offset)

            response = await client.market_data.get_last_trades(
                figi=instrument["figi"],
                from_=interval_from,
                to=interval_to
            )

            print(f"Fetched {len(response.trades)} trades")
            # Output: Fetched 15847 trades

            if not response.trades:
                break

            # Process and append trades
            for trade in response.trades:
                trade_data = pd.DataFrame([{
                    "price": float(trade.price.units + trade.price.nano / 1e9),
                    "quantity": trade.quantity,
                    "time": trade.time,
                    "direction": trade.direction
                }])
                history_df = pd.concat([history_df, trade_data])

            time_offset += ONE_HOUR_TO_MINUTES

        # Merge with existing data file
        existing_df = pd.read_csv("./data/SBER_trades.csv")
        existing_df["time"] = pd.to_datetime(existing_df["time"], utc=True)

        # Remove duplicates and sort
        merged_df = merge_two_frames(existing_df, history_df)
        print(f"Total trades after merge: {len(merged_df)}")
        # Output: Total trades after merge: 125634

asyncio.run(sync_and_backfill())

```

--------------------------------

### Order Domain Model for Lifecycle Tracking (Python)

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Defines a data structure for representing a trading order, including its full lifecycle from creation to closure. It stores essential order details such as instrument, quantity, direction, and price points (open, stop, take). The model supports tracking the order's status, calculating profit/loss, determining win/loss status, and can be converted to/from a dictionary for data persistence (e.g., CSV).

```python
from domains.order import Order
from tinkoff.invest import OrderDirection
from datetime import datetime

# Create new order
order = Order(
    id="ord_001",
    group_id="grp_1",
    instrument="SBER",
    open=150.50,
    stop=148.00,
    take=154.50,
    quantity=2,
    direction=OrderDirection.ORDER_DIRECTION_BUY.value,
    time=datetime(2024, 1, 15, 10, 30, 0)
)

# Order lifecycle
print(order.status)  # "active"
print(order.result)  # 0 (not closed yet)

# When closed at take-profit
order.status = "close"
order.close = 154.50
if order.direction == OrderDirection.ORDER_DIRECTION_BUY.value:
    order.result = order.close - order.open  # 154.50 - 150.50 = 4.0
    order.is_win = order.result > 0  # True

print(f"Result: {order.result} points")  # Result: 4.0 points
print(f"Win: {order.is_win}")  # Win: True

# Convert to dict for CSV persistence
order_dict = dict(order)
# {"id": "ord_001", "group_id": "grp_1", "instrument": "SBER", ...}

# Load from CSV data
csv_row = {
    "id": "ord_001",
    "open": "150.50",
    "close": "154.50",
    "stop": "148.00",
    "take": "154.50",
    "quantity": "2",
    "direction": "1",
    "time": "2024-01-15 10:30:00",
    "status": "close",
    "result": "4.0",
    "is_win": "True"
}
loaded_order = Order.from_dict(csv_row)

```

--------------------------------

### ProfileTouchStrategy for Trade Signal Generation (Python)

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Implements a strategy that detects when price returns to high-volume price levels and generates trade signals based on candle confirmation. It analyzes trade data, builds volume clusters, monitors price proximity to volume levels, and checks for confirmation candles before returning potential Order objects with entry, stop, and take-profit levels. Requires pandas for data manipulation and tinkoff.invest for order types.

```python
from strategies.profile_touch_strategy import ProfileTouchStrategy
from tinkoff.invest import OrderDirection
import pandas as pd

# Initialize strategy for an instrument
strategy = ProfileTouchStrategy(instrument_name="SBER")
strategy.start()

# Process incoming trade data
trade_df = pd.DataFrame([{
    "figi": "BBG004730N88",
    "direction": 1,  # TradeDirection.TRADE_DIRECTION_BUY
    "price": 150.50,
    "quantity": 100,
    "time": pd.Timestamp.now(tz='UTC')
}])

# Analyze returns None or List[Order]
orders = strategy.analyze(trade_df)

if orders:
    for order in orders:
        print(f"Signal: {order.direction}")
        print(f"Entry: {order.open}")
        print(f"Stop: {order.stop}")
        print(f"Target: {order.take}")
        # Output example:
        # Signal: OrderDirection.ORDER_DIRECTION_BUY
        # Entry: 150.50
        # Stop: 148.98
        # Target: 154.56

# Strategy logic:
# 1. Builds volume clusters every hour
# 2. Monitors when price touches volume levels (within 3% range)
# 3. Waits 90+ minutes after level formation for first touch
# 4. Checks 5-min signal candle for confirmation:
#    - Long: Buyers dominate AND max volume in lower part
#    - Short: Sellers dominate AND max volume in upper part
# 5. Returns Order objects with entry, stop, and take-profit

```

--------------------------------

### Calculate Backtest Results

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Calculates backtesting results from a list of collected orders. It determines the number of winning and losing trades, calculates the win rate, and sums the total profit.

```python
# Calculate backtest results
winning_trades = [o for o in all_orders if o.is_win]
losing_trades = [o for o in all_orders if not o.is_win]

total_trades = len(all_orders)
win_rate = len(winning_trades) / total_trades * 100
total_profit = sum(o.result for o in all_orders)

print(f"\nBacktest Results:")
print(f"Total trades: {total_trades}")
print(f"Win rate: {win_rate:.2f}%")
print(f"Total profit: {total_profit:.2f} points")
```

--------------------------------

### Test Trading Strategy on Historical Data

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Backtests a volume profile trading strategy on historical trade data and includes visualization capabilities. It loads trade data, initializes the strategy, and feeds the historical data to build the initial state for testing. Requires pandas for data manipulation and specific strategy modules.

```python
from strategies.profile_touch_strategy import ProfileTouchStrategy
from utils.strategy_util import create_empty_df
import pandas as pd

# Load historical trade data
df = pd.read_csv("./data/trades_SBER_2022-05-04.csv")
df["time"] = pd.to_datetime(df["time"], utc=True)
print(f"Loaded {len(df)} trades")

# Initialize strategy
strategy = ProfileTouchStrategy(instrument_name="SBER")
strategy.start()

# Feed entire history to build initial state
strategy.set_df(df)

```

--------------------------------

### Process Trades Sequentially and Collect Orders

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Iterates through a DataFrame of trades, analyzes each trade using a strategy, and collects generated orders. It prints entry details for each order and extends a list of all orders.

```python
all_orders = []
for idx, row in df.iterrows():
    trade_df = pd.DataFrame([row])
    orders = strategy.analyze(trade_df)

    if orders:
        all_orders.extend(orders)
        for order in orders:
            print(f"Entry at {order.time}: {order.direction} @ {order.open}")
            print(f"  Stop: {order.stop}, Target: {order.take}")
```

--------------------------------

### Aggregate OHLCV Candles in Python

Source: https://github.com/tromario/tinkoff-invest-volume-analysis-robot/blob/master/jupyter/project.ipynb

This script reads trading data, converts it into OHLCV (Open, High, Low, Close, Volume) candles aggregated per minute, and determines the direction of each candle. It uses pandas for data manipulation and numpy for numerical operations. The output includes time, open, close, high, low, total volume, candle direction, and the price with maximum volume.

```python
import numpy as np

# Set pandas display option for maximum rows
pd.set_option("max_rows", None)

# Load and preprocess the input DataFrame
df1 = pandas.read_csv('./../data/USD000UTSTOM-20220512.csv', sep=',')
df1.time = pd.to_datetime(df1.time)
df1.price = pd.to_numeric(df1.price)
df1.quantity = pd.to_numeric(df1.quantity)

# Define a function to aggregate OHLCV data for a group of trades
def agg_ohlcv(x):
    if x.empty:
        return pd.Series({
            'low': 0,
            'high': 0,
            'open': 0,
            'close': 0,
            'total_volume': 0,
            'max_volume_price': 0
        })
    
    price = x['price'].values
    quantity = x['quantity'].values
    return pd.Series({
        'low': min(price),
        'high': max(price),
        'open': price[0],
        'close': price[-1],
        'total_volume': sum(quantity),
        'max_volume_price': x.groupby(['price'])[['quantity']].sum().idxmax()[0]
    })

# Resample data into 1-minute candles and forward fill missing values
candles = df1.set_index(['time'])
candles = candles.resample('1min').apply(agg_ohlcv)
candles = candles.ffill()

# Determine candle direction (doji, bullish, bearish)
candles.loc[candles['close'] == candles['open'], 'direction'] = 0 # doji candle
candles.loc[candles['close'] > candles['open'], 'direction'] = 1 # bullish candle
candles.loc[candles['open'] > candles['close'], 'direction'] = 2 # bearish candle
candles['time'] = candles.index

# Select and reorder columns for the final output
candles = candles[['time', 'open', 'close', 'high', 'low', 'total_volume', 'direction', 'max_volume_price']]
candles = candles.reset_index(drop=True)

# Optional: display the candles DataFrame
# candles
```

--------------------------------

### OrderService Trade Management with Tinkoff API (Python)

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Manages the lifecycle of trading orders, including creation, monitoring, and execution via the Tinkoff Invest API. It supports real-time trade execution, Telegram notifications for alerts, and end-of-day statistics generation. The service requires order details and current market prices as input.

```python
from services.order_service import OrderService
from domains.order import Order
from tinkoff.invest import OrderDirection
from datetime import datetime

# Initialize order service
order_service = OrderService(
    is_notification=True,  # Send Telegram alerts
    can_open_orders=True   # Execute real trades
)
order_service.start()

# Create a new order
order = Order(
    id="order_123",
    group_id="group_1",
    instrument="SBER",
    open=150.50,
    stop=148.00,
    take=154.50,
    quantity=2,
    direction=OrderDirection.ORDER_DIRECTION_BUY.value,
    time=datetime.now()
)

order_service.create_order(order)
# Logs: ✅ ТВ SBER: цена 150.50, тейк 154.50, стоп 148.00
# Sends Telegram notification if configured
# Executes market order via Tinkoff API

# Monitor positions (called continuously with incoming trades)
order_service.processed_orders(
    instrument="SBER",
    current_price=152.00,
    time=datetime.now()
)
# Automatically closes if price hits stop or target

# Generate statistics at end of trading day
order_service.write_statistics()
# Output to logs/statistics-SBER.log:
# количество сделок: 10
# успешных сделок: 7
# заработано пунктов: 25.5
# отрицательных сделок: 3
# потеряно пунктов: -7.2
# итого пунктов: 18.3

```

--------------------------------

### Volume Cluster Calculation from Tick Data (Python)

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Converts tick data into OHLC (Open, High, Close, Low) candles and identifies the price with maximum volume within each candle. It also calculates the buyer/seller ratio and provides winning condition analysis. This utility requires tick data as a pandas DataFrame and outputs processed candle data.

```python
from utils.strategy_util import ticks_to_cluster, calculate_ratio
import pandas as pd
from tinkoff.invest import TradeDirection

# Sample tick data
ticks_df = pd.DataFrame({
    "figi": ["BBG004730N88"] * 10,
    "direction": [TradeDirection.TRADE_DIRECTION_BUY.value] * 10,
    "price": [150.1, 150.2, 150.15, 150.3, 150.25, 150.2, 150.35, 150.3, 150.4, 150.38],
    "quantity": [100, 200, 150, 300, 250, 180, 220, 190, 210, 170],
    "time": pd.date_range("2024-01-15 10:00:00", periods=10, freq="10s", tz="UTC")
})

# Convert ticks to 1-minute candles with max volume price
candles = ticks_to_cluster(ticks_df, period="1min")
print(candles)
# Output columns: time, open, close, high, low, total_volume, direction, max_volume_price
# max_volume_price: 150.3 (price with highest cumulative volume)

# Calculate buyer/seller ratio
candles_with_ratio = calculate_ratio(candles)
print(candles_with_ratio[["long", "short", "percent", "win"]])
# long: 65.2%  (buyers dominate)
# short: 34.8% (sellers)
# percent: 35% (max volume is 35% from low)
# win: True (long signal confirmed: long>50% AND percent<=40%)

# Winning conditions:
# Long: long > 50% AND max_volume in lower 40% of candle
# Short: short > 50% AND max_volume in upper 40% of candle

```

--------------------------------

### Send Telegram Trade Alerts and Notifications

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Utilizes the TelegramService to send trade alerts and notifications via a Telegram bot. Requires a bot token and chat ID for initialization. Supports Markdown formatting and disables link previews. Users must have initiated a chat with the bot to receive messages.

```python
from services.telegram_service import TelegramService

# Initialize with bot credentials
telegram = TelegramService(
    bot_token="1234567890:ABCdefGHIjklMNOpqrsTUVwxyz",
    chat_id="987654321"
)

# Send trade notification
response = telegram.post("✅ ТВ SBER: цена 150.50, тейк 154.50, стоп 148.00")

if response:
    print(response)
    # Output: {'ok': True, 'result': {...message_id, date, text...}}

# Send position closure alert
telegram.post("закрыта позиция на SBER: результат 4.5")

# Notification scenarios:
# - New position opened
# - Position closed (take-profit or stop-loss)
# - End of trading day statistics
# - Error conditions or warnings

# Message format uses Markdown and disables link previews
# Requires pre-configured Telegram bot via @BotFather
# User must have started chat with bot to receive messages
```

--------------------------------

### Validate Exchange Trading Hours and Market Status

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Provides utility functions to validate trading hours and market status for exchange operations. Functions check if the exchange is open, if order placement is allowed, and if the current time falls within the premarket period. These checks help in managing trading activities around market open and close times.

```python
from utils.exchange_util import is_open_exchange, is_open_orders, is_premarket_time
from datetime import datetime
import pytz

# Check if exchange is currently open
if is_open_exchange():
    print("Exchange is open")
else:
    print("Exchange is closed")
# Returns False after 16:00 MSK

# Check if order placement is allowed
moscow_time = datetime.now(pytz.timezone('Europe/Moscow'))
can_trade = is_open_orders(moscow_time)
print(can_trade)
# Returns False after 15:00 MSK
# Gives 1 hour buffer before exchange close

# Check if in premarket period
time = datetime(2024, 1, 15, 6, 30, 0, tzinfo=pytz.UTC)
if is_premarket_time(time):
    print("Premarket - skip analysis")
else:
    print("Regular hours - analyze trades")
# Returns True before 07:00 MSK
# Prevents trading during low-liquidity premarket

# Use cases:
# - Close all positions before exchange close
# - Prevent order entry too close to session end
# - Skip analysis during premarket volatility
```

--------------------------------

### Merge and Sort DataFrames in Python

Source: https://github.com/tromario/tinkoff-invest-volume-analysis-robot/blob/master/jupyter/project.ipynb

This script merges two pandas DataFrames read from CSV files, specifically handling time-series data. It ensures chronological order and saves the combined data to a new CSV file. Dependencies include the pandas library.

```python
import pandas

# Load and process the first CSV file
new_df = pandas.read_csv('./../data/temp/SBER-20220506.csv', sep=',')
new_df.time = pd.to_datetime(new_df.time)

# Load and process the second CSV file
source_df = pandas.read_csv('./../data/SBER-20220506.csv', sep=',')
source_df.time = pd.to_datetime(source_df.time)

# Determine time range for merging
first_time = new_df.iloc[0]['time']
last_time = new_df.iloc[-1]['time']
search_condition = (source_df['time'] >= first_time) & (source_df['time'] <= last_time)

# Merge, sort, and save the result
result_df = source_df.drop(source_df.loc[search_condition].index)
result_df = pd.concat([result_df, new_df]).rename_axis('index')
result_df = result_df.sort_values(['time', 'index']).reset_index(drop=True)

result_df.to_csv('./../data/SBER-20220506-merge-test.csv', mode='a', header=True, index=False)
```

--------------------------------

### Price Range Validation Against Volume Level (Python)

Source: https://context7.com/tromario/tinkoff-invest-volume-analysis-robot/llms.txt

Checks if a current trading price falls within an acceptable percentage range of a specified volume level. This function is useful for validating price deviations from key volume clusters, accounting for spread and tick size variations. It takes the current price and volume level as input and returns a boolean indicating if the price is within the defined range.

```python
from utils.strategy_util import is_price_in_range_cluster

# Volume level identified at 150.00
volume_level = 150.00
current_price = 150.40

# Check if price is within 3% range (PERCENTAGE_VOLUME_LEVEL_RANGE)
is_valid = is_price_in_range_cluster(current_price, volume_level)
print(is_valid)
# Output: True

# Calculation:
# level_range = 150.00 * 0.03 / 100 = 0.045
# increased_level = 150.00 + 0.045 = 150.045
# reduced_level = 150.00 - 0.045 = 149.955
# is_valid = 149.955 <= 150.40 <= 150.045 = True

# Use case: Allow slight price deviation from exact level
# Accounts for spread and tick size variations

```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.