أدوات الوكيل لتحليل المحافظ الاستثمارية

تحليل المحفظة بوصفه أدوات للوكيل

حساب العوائد المئوية

import pandas as pd
import yfinance as yf

def calculate_returns(ticker: str, period: str = '1y') -> dict:
    hist = yf.Ticker(ticker).history(period=period)['Close']
    if hist.empty:
        return {'error': f'No data for {ticker}'}

    daily_returns = hist.pct_change().dropna()
    cumulative    = (1 + daily_returns).cumprod() - 1

    return {
        'ticker':            ticker,
        'period':            period,
        'total_return_pct':  round(float(cumulative.iloc[-1]) * 100, 2),
        'avg_daily_return':  round(float(daily_returns.mean()) * 100, 4),
        'best_day_pct':      round(float(daily_returns.max()) * 100, 2),
        'worst_day_pct':     round(float(daily_returns.min()) * 100, 2)
    }

نسبة شارب

import numpy as np
import yfinance as yf

def sharpe_ratio(ticker: str, period: str = '1y',
                 risk_free_annual: float = 0.05) -> dict:
    hist = yf.Ticker(ticker).history(period=period)['Close']
    daily_returns = hist.pct_change().dropna()

    risk_free_daily = risk_free_annual / 252  # annualized to daily
    excess_returns  = daily_returns - risk_free_daily

    if daily_returns.std() == 0:
        return {'error': 'Zero volatility — cannot compute Sharpe ratio'}

    sharpe = float(excess_returns.mean() / excess_returns.std() * np.sqrt(252))
    return {
        'ticker': ticker,
        'sharpe_ratio': round(sharpe, 3),
        'interpretation': 'excellent' if sharpe > 2 else 'good' if sharpe > 1 else 'acceptable' if sharpe > 0 else 'poor'
    }

أقصى تراجع

import yfinance as yf
import numpy as np

def max_drawdown(ticker: str, period: str = '1y') -> dict:
    hist = yf.Ticker(ticker).history(period=period)['Close']
    if hist.empty:
        return {'error': 'No data'}

    rolling_max = hist.cummax()
    drawdown    = (hist - rolling_max) / rolling_max
    mdd         = float(drawdown.min())
    mdd_date    = str(drawdown.idxmin().date())

    return {
        'ticker':           ticker,
        'max_drawdown_pct': round(mdd * 100, 2),
        'drawdown_date':    mdd_date,
        'interpretation':   'severe' if mdd < -0.30 else 'moderate' if mdd < -0.15 else 'mild'
    }

مصفوفة الارتباط

import pandas as pd
import yfinance as yf

def correlation_matrix(tickers: list[str], period: str = '1y') -> dict:
    prices = yf.download(tickers, period=period)['Close']
    returns = prices.pct_change().dropna()
    corr = returns.corr().round(3)

    # Convert to JSON-serializable format
    result = {}
    for t1 in tickers:
        result[t1] = {}
        for t2 in tickers:
            if t1 in corr.columns and t2 in corr.index:
                result[t1][t2] = float(corr.loc[t2, t1])
    return {'correlation': result, 'period': period}

التقلب وبيتا

import numpy as np
import yfinance as yf

def volatility_and_beta(ticker: str, benchmark: str = 'SPY',
                        period: str = '1y') -> dict:
    data = yf.download([ticker, benchmark], period=period)['Close'].pct_change().dropna()

    if ticker not in data.columns or benchmark not in data.columns:
        return {'error': 'Insufficient data'}

    t_ret = data[ticker]
    b_ret = data[benchmark]

    vol_annualized = float(t_ret.std() * np.sqrt(252))

    cov = np.cov(t_ret, b_ret)
    beta = float(cov[0][1] / cov[1][1]) if cov[1][1] != 0 else 0.0

    return {
        'ticker':          ticker,
        'annual_vol_pct':  round(vol_annualized * 100, 2),
        'beta':            round(beta, 3),
        'market_sensitivity': 'high' if beta > 1.3 else 'market' if beta > 0.7 else 'low'
    }

العائد والمخاطر على مستوى المحفظة

import pandas as pd, numpy as np
import yfinance as yf

def portfolio_metrics(holdings: dict, period: str = '1y') -> dict:
    tickers  = list(holdings.keys())
    weights  = np.array(list(holdings.values()))
    weights /= weights.sum()  # normalize to 1.0

    prices  = yf.download(tickers, period=period)['Close'].pct_change().dropna()
    port_ret = (prices * weights).sum(axis=1)
    cov_matrix = prices.cov() * 252  # annualized

    port_return  = float(port_ret.mean() * 252)
    port_vol     = float(np.sqrt(weights @ cov_matrix.values @ weights))
    port_sharpe  = port_return / port_vol if port_vol > 0 else 0

    return {
        'annual_return_pct':  round(port_return * 100, 2),
        'annual_vol_pct':     round(port_vol * 100, 2),
        'sharpe_ratio':       round(port_sharpe, 3),
        'holdings':           {t: round(float(w), 3) for t, w in zip(tickers, weights)}
    }

اقتراح إعادة الموازنة

def rebalancing_suggestion(current_values: dict, target_weights: dict) -> dict:
    total_value = sum(current_values.values())
    current_weights = {t: v / total_value for t, v in current_values.items()}
    trades = {}

    for ticker in target_weights:
        current_w = current_weights.get(ticker, 0)
        target_w  = target_weights[ticker]
        diff_w    = target_w - current_w
        diff_usd  = diff_w * total_value

        if abs(diff_w) > 0.01:  # only suggest if drift > 1%
            trades[ticker] = {
                'action':      'BUY' if diff_usd > 0 else 'SELL',
                'amount_usd':  round(abs(diff_usd), 2),
                'from_weight': round(current_w, 4),
                'to_weight':   round(target_w, 4)
            }
    return {'total_portfolio': round(total_value, 2), 'trades': trades}

if __name__ == '__main__':
    current = {'AAPL': 6000, 'BND': 2000, 'VTI': 2000}
    target = {'AAPL': 0.4, 'BND': 0.3, 'VTI': 0.3}
    result = rebalancing_suggestion(current, target)
    print('Total portfolio: $%s' % result['total_portfolio'])
    for ticker, trade in result['trades'].items():
        print(f"  {ticker}: {trade['action']} ${trade['amount_usd']} "
              f"(weight {trade['from_weight']} -> {trade['to_weight']})")

تسجيل أدوات التحليل لدى LLM

FINANCIAL_TOOLS = [
    {
        'type': 'function',
        'function': {
            'name': 'calculate_returns',
            'description': 'Calculate total and daily returns for a ticker.',
            'parameters': {'type': 'object',
                           'properties': {
                               'ticker': {'type': 'string'},
                               'period': {'type': 'string', 'default': '1y'}
                           }, 'required': ['ticker']}
        }
    },
    {
        'type': 'function',
        'function': {
            'name': 'sharpe_ratio',
            'description': 'Compute the annualized Sharpe ratio.',
            'parameters': {'type': 'object',
                           'properties': {
                               'ticker': {'type': 'string'},
                               'period': {'type': 'string', 'default': '1y'}
                           }, 'required': ['ticker']}
        }
    }
]

if __name__ == '__main__':
    print('Registered financial tools for the LLM:')
    for tool in FINANCIAL_TOOLS:
        fn = tool['function']
        print(f"  {fn['name']}: {fn['description']}")

إنشاء ملخص للمحفظة بلغة واضحة

import openai, json

client = openai.OpenAI(api_key='YOUR_OPENAI_KEY')

def portfolio_narrative(metrics: dict, holdings: dict) -> str:
    prompt = (
        f'Portfolio metrics:\n{json.dumps(metrics, indent=2)}\n\n'
        f'Holdings: {json.dumps(holdings)}\n\n'
        f'Write a 3-sentence plain-language portfolio assessment for a retail investor.\n'
        f'Include: overall performance, main risk, and one recommendation.'
    )
    resp = client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role': 'user', 'content': prompt}]
    )
    return resp.choices[0].message.content

التخزين المؤقت للمقاييس المحسوبة

import time, hashlib, json

metrics_cache = {}

def cached_metric(func, *args, ttl: int = 3600):
    key = hashlib.md5(f'{func.__name__}:{args}'.encode()).hexdigest()
    entry = metrics_cache.get(key)
    if entry and time.time() - entry['ts'] < ttl:
        return entry['data']
    result = func(*args)
    metrics_cache[key] = {'data': result, 'ts': time.time()}
    return result

# Usage:
# sharpe = cached_metric(sharpe_ratio, 'AAPL', '1y')

if __name__ == '__main__':
    def slow_average(*nums):
        return sum(nums) / len(nums)

    r1 = cached_metric(slow_average, 1, 2, 3)
    r2 = cached_metric(slow_average, 1, 2, 3)
    print('First call result:', r1)
    print('Second call result (served from cache):', r2)
    print('Cache entries:', len(metrics_cache))

ماذا تشير نسبة شارب التي تتجاوز 2.0؟

مراجعة أدوات تحليل المحفظة