Reports

from nautilus_trader.analysis import ReportProvider

# From a completed backtest run
orders_report = engine.generate_orders_report()

# Or from any cache, such as a live node's
orders_report = ReportProvider.generate_orders_report(cache.orders())

# From a completed backtest run
fills_report = engine.generate_order_fills_report()

# Or from any cache
fills_report = ReportProvider.generate_order_fills_report(cache.orders())

# From a completed backtest run
fills_report = engine.generate_fills_report()

# Or from any cache
fills_report = ReportProvider.generate_fills_report(cache.orders())

# From a completed backtest run, which includes snapshots automatically
positions_report = engine.generate_positions_report()

# Or from any cache
positions_report = ReportProvider.generate_positions_report(
    positions=cache.positions(),
    snapshots=cache.position_snapshots(),  # Needed for NETTING OMS totals
)

from nautilus_trader.model import Venue

venue = Venue("BINANCE")

# From a completed backtest run
account_report = engine.generate_account_report(venue=venue)

# Or from any cache
account_report = ReportProvider.generate_account_report(cache.account_for_venue(venue))

from decimal import Decimal

# Accessing PnL across positions
for position in cache.positions_closed():
    realized = position.realized_pnl  # Money in the position's cost currency, or None

    if realized is None or realized.currency == base_currency:
        continue

    # Converting by hand: cache.get_xrate() returns a float, so wrap rates as Decimal
    rate = Decimal(str(my_fx_rates[(realized.currency, base_currency)]))
    converted = realized.as_decimal() * rate

from decimal import Decimal

from nautilus_trader.model import Money

pnl_by_currency = {}

for position in cache.positions(instrument_id=instrument_id):
    # Archived cycles are stored under the live position's ID
    snapshots = cache.position_snapshots(position_id=position.id)

    for pnl in (position.realized_pnl, *(s.realized_pnl for s in snapshots)):
        if pnl is None:
            continue

        running = pnl_by_currency.get(pnl.currency, Decimal(0))
        pnl_by_currency[pnl.currency] = running + pnl.as_decimal()

# Create Money objects for each currency
total_pnls = [Money.from_decimal(amount, currency) for currency, amount in pnl_by_currency.items()]

# After backtest run
engine.run()

# Access result statistics
result = engine.get_result()

# Generate reports from the backtest engine
fills_report = engine.generate_fills_report()
venue = engine.list_venues()[0]
account_report = engine.generate_account_report(venue=venue)

# Or access data directly for custom analysis
orders = engine.cache.orders()
positions = engine.cache.positions()
snapshots = engine.cache.position_snapshots()

# Access backtest result statistics
result = engine.get_result()

# Get different categories of statistics
stats_pnls = result.stats_pnls  # Keyed by currency code, then statistic name
stats_returns = result.stats_returns  # Keyed by statistic name
stats_general = result.stats_general  # Keyed by statistic name

from nautilus_trader.analysis import create_tearsheet

# After backtest run
engine.run()

# Generate interactive HTML tearsheet
create_tearsheet(engine, output_path="tearsheet.html")

import pandas as pd

from nautilus_trader.analysis import create_equity_curve

returns = pd.Series(
    [0.01, -0.005, 0.002],
    index=pd.date_range("2024-01-01", periods=3, tz="UTC"),
)
fig = create_equity_curve(returns, title="My Strategy Equity")
fig.show()  # Display in browser
fig.write_image("equity.png")  # Export to PNG (requires kaleido)

uv pip install "nautilus_trader[visualization]"

from datetime import timedelta

from nautilus_trader.analysis import ReportProvider
from nautilus_trader.common import DataActor
from nautilus_trader.common import TimeEvent


class ReportingActor(DataActor):
    def on_start(self) -> None:
        # Schedule periodic reporting
        self.clock.set_timer(
            name="generate_reports",
            interval=timedelta(minutes=30),
            callback=self.generate_reports,
        )

    def generate_reports(self, event: TimeEvent) -> None:
        # Generate and log reports
        positions_report = ReportProvider.generate_positions_report(
            positions=self.cache.positions(),
            snapshots=self.cache.position_snapshots(),
        )

        # Save or transmit report
        positions_report.to_csv(f"positions_{event.ts_event}.csv")

import pandas as pd

# Run the backtest
engine.run()

# Collect results
positions_closed = engine.cache.positions_closed()
result = engine.get_result()
stats_pnls = result.stats_pnls
stats_returns = result.stats_returns
stats_general = result.stats_general

# Create summary dictionary
results = {
    "total_positions": len(positions_closed),
    "pnl_total": stats_pnls.get("USD", {}).get("PnL (total)"),
    "win_rate": stats_pnls.get("USD", {}).get("Win Rate"),
    "sharpe_ratio": stats_returns.get("Sharpe Ratio (252 days)"),
    "profit_factor": stats_returns.get("Profit Factor"),
    "long_ratio": stats_general.get("Long Ratio"),
}

# Display results
results_df = pd.DataFrame([results])
print(results_df.T)  # Transpose for vertical display