main.py - master - SimpleSpace/AutoInvestDataSync

main.py
236 строк8 KB
Alex Just
Start commit for mvp 0.0.5 version
04 май 2026, 22:47
04 май 2026, 22:47e12863e
from __future__ import annotations

import logging
import os
import sys
from datetime import datetime
from pathlib import Path

import click
import yaml
from dotenv import load_dotenv

load_dotenv()

CONFIG_PATH = Path(__file__).parent / "config.yaml"


def _load_config() -> dict:
    if CONFIG_PATH.exists():
        with open(CONFIG_PATH, "r", encoding="utf-8") as f:
            return yaml.safe_load(f) or {}
    return {}


def _get_token() -> str:
    token = os.environ.get("T_INVEST_TOKEN", "")
    if not token or token == "your_token_here":
        click.echo("Error: T_INVEST_TOKEN not set. Add it to .env file.", err=True)
        sys.exit(1)
    return token


@click.group()
def cli():
    """AutoInvestDataSync — SilverFir Investment Report"""
    pass


@cli.command()
@click.option("--no-open", is_flag=True, help="Do not open the file in Numbers")
@click.option("--output", "-o", default=None, help="Output file path")
@click.option("--verbose", "-v", is_flag=True, help="Verbose logging")
def sync(no_open: bool, output: str, verbose: bool):
    """Fetch data from T-Invest API and generate XLSX report."""
    logging.basicConfig(
        level=logging.DEBUG if verbose else logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )
    logger = logging.getLogger(__name__)

    config = _load_config()
    token = _get_token()

    click.echo("Connecting to T-Invest API...")
    from sync.client import TInvestClient
    from sync.portfolio import fetch_portfolio
    from sync.instruments import enrich_positions
    from sync.dividends import enrich_payouts
    from sync.operations import fetch_operations
    from sync.analytics import calculate_analytics
    from sync.history import add_history_entry
    from sync.macro import fetch_macro_data, fetch_market_index
    from export.xlsx_builder import build_xlsx
    from export.numbers_bridge import open_in_numbers

    with TInvestClient(token=token) as tinvest:
        try:
            accounts = tinvest.get_accounts()
            if not accounts:
                click.echo("No open accounts found.")
                return

            click.echo(f"Found {len(accounts)} account(s)")

            click.echo("Fetching portfolio positions...")
            all_portfolios = fetch_portfolio(tinvest, accounts)

            all_positions = []
            for ap in all_portfolios:
                all_positions.extend(ap.positions)

            click.echo(f"Total positions: {len(all_positions)}")

            click.echo("Enriching instrument metadata...")
            enrich_positions(tinvest, all_positions)

            click.echo("Fetching dividends and coupons...")
            payouts = enrich_payouts(tinvest, all_positions)

            click.echo("Fetching operations history...")
            purchases, first_buy_dates = fetch_operations(tinvest, accounts)

            click.echo("Calculating analytics...")
            analytics = calculate_analytics(all_positions, first_buy_dates)

            click.echo("Fetching macro data (inflation)...")
            macro = fetch_macro_data()

            click.echo("Fetching market index (EQMX)...")
            market_index = fetch_market_index(tinvest)
            macro.market_index = market_index

            click.echo("Updating capital history...")
            history_file = config.get("history_file", "capital_history.yaml")
            history = add_history_entry(
                history_file,
                analytics["total_buy"],
                analytics["total_current"],
            )

            if output:
                output_path = output
            else:
                output_dir = os.path.expanduser(config.get("output_dir", "~/Desktop"))
                date_str = datetime.now().strftime("%Y-%m-%d")
                output_path = os.path.join(output_dir, f"SilverFir_Report_{date_str}.xlsx")

            click.echo(f"Generating XLSX: {output_path}")
            build_xlsx(all_portfolios, analytics, purchases, payouts, history, output_path, macro)

            if not no_open and config.get("open_in_numbers", True):
                click.echo("Opening in Numbers...")
                open_in_numbers(output_path)

            click.echo(f"\nDone! Report saved to: {output_path}")
            click.echo(f"  Total portfolio: {analytics['total_current']:,.2f} RUB")
            click.echo(f"  P&L: {analytics['total_pnl']:+,.2f} RUB")
            if macro.inflation_imf_pct:
                click.echo(f"  Inflation (IMF {macro.inflation_imf_year}): {macro.inflation_imf_pct:.1f}%")
            if macro.inflation_cpi_wb_pct:
                click.echo(f"  Inflation CPI (WB {macro.inflation_cpi_wb_year}): {macro.inflation_cpi_wb_pct:.1f}%")
            if macro.market_index.change_year_pct:
                mi = macro.market_index
                click.echo(f"  Market index (EQMX): {mi.current_price:.2f} RUB ({mi.change_year_pct:+.2f}% YoY)")

        except Exception as e:
            logger.exception("Sync failed")
            click.echo(f"Error: {e}", err=True)
            sys.exit(1)


@cli.command()
@click.option("--time", "schedule_time", default="20:00", help="Schedule time HH:MM")
def schedule(schedule_time: str):
    """Install launchd schedule for daily sync."""
    import subprocess

    python_path = sys.executable
    script_path = os.path.abspath(__file__)
    project_dir = os.path.dirname(os.path.abspath(__file__))

    hour, minute = schedule_time.split(":")

    plist_content = f'''<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.autoinvest.sync</string>
    <key>ProgramArguments</key>
    <array>
        <string>{python_path}</string>
        <string>{script_path}</string>
        <string>sync</string>
    </array>
    <key>WorkingDirectory</key>
    <string>{project_dir}</string>
    <key>StartCalendarInterval</key>
    <dict>
        <key>Hour</key>
        <integer>{hour}</integer>
        <key>Minute</key>
        <integer>{minute}</integer>
    </dict>
    <key>StandardOutPath</key>
    <string>/tmp/autoinvest.log</string>
    <key>StandardErrorPath</key>
    <string>/tmp/autoinvest.err</string>
</dict>
</plist>
'''

    plist_path = os.path.expanduser("~/Library/LaunchAgents/com.autoinvest.sync.plist")
    os.makedirs(os.path.dirname(plist_path), exist_ok=True)

    with open(plist_path, "w", encoding="utf-8") as f:
        f.write(plist_content)

    subprocess.run(["launchctl", "unload", plist_path], capture_output=True)
    subprocess.run(["launchctl", "load", plist_path], check=True)

    click.echo(f"Schedule installed: daily at {schedule_time}")
    click.echo(f"plist: {plist_path}")


@cli.command()
def unschedule():
    """Remove launchd schedule."""
    import subprocess

    plist_path = os.path.expanduser("~/Library/LaunchAgents/com.autoinvest.sync.plist")
    if os.path.exists(plist_path):
        subprocess.run(["launchctl", "unload", plist_path], capture_output=True)
        os.remove(plist_path)
        click.echo("Schedule removed.")
    else:
        click.echo("No schedule found.")


@cli.command()
def status():
    """Show current status and configuration."""
    config = _load_config()
    click.echo("Configuration:")
    click.echo(f"  Output dir: {config.get('output_dir', '~/Desktop')}")
    click.echo(f"  Open in Numbers: {config.get('open_in_numbers', True)}")
    click.echo(f"  Schedule time: {config.get('schedule_time', '20:00')}")

    token = os.environ.get("T_INVEST_TOKEN", "")
    if token and token != "your_token_here":
        click.echo(f"  Token: {'*' * 8}{token[-4:]}")
    else:
        click.echo("  Token: NOT SET")

    history_file = config.get("history_file", "capital_history.yaml")
    if os.path.exists(history_file):
        from sync.history import load_history
        history = load_history(history_file)
        if history:
            last = history[-1]
            click.echo(f"\nLast sync: {last.get('date', 'unknown')}")
            click.echo(f"  Portfolio: {last.get('current_value', 0):,.2f} RUB")


if __name__ == "__main__":
    cli()