sevenhill/TradeBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add fractional Kelly position sizing

Browse Source
This commit is contained in:
Codex
2026-06-20 22:22:19 +03:00
parent ccf457481b
commit 7bbb721da1
8 changed files with 147 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crypto_spot_bot/bot.py
+3 -2
View File
@@ -146,6 +146,7 @@ class CryptoSpotBot:
forecast = self.market.forecasts.get(symbol, {})
learning = self.learner.adjustment_for(symbol, str(pattern.get("label", ""))).as_dict()
learning["adaptive_rules"] = self._with_exposure_context(learning.get("adaptive_rules") or {})
account = self.broker.account_state(prices)
llm = {}
if (
self.settings.llm_advisor_enabled
@@ -162,10 +163,10 @@ class CryptoSpotBot:
pattern=pattern,
learning=learning,
open_positions_for_symbol=open_count,
account=self.broker.account_state(prices),
account=account,
)
).as_dict()
signal = self.strategy.entry_signal(symbol, candles, ticker, open_count, pattern, learning, llm, forecast)
signal = self.strategy.entry_signal(symbol, candles, ticker, open_count, pattern, learning, llm, forecast, account)
self.storage.insert_signal(signal)
if signal.action == "BUY" and ticker is not None:
self.broker.buy(
crypto_spot_bot/config.py
+6
View File
@@ -92,6 +92,9 @@ class Settings:
rebound_entry_confidence: float
rebound_min_probability: float
rebound_max_position_usdt: float
kelly_sizing_enabled: bool
kelly_fraction: float
kelly_max_fraction: float
time_series_forecast_enabled: bool
time_series_min_candles: int
time_series_validation_window: int
@@ -212,6 +215,9 @@ def load_settings(env_file: str | Path | None = None) -> Settings:
rebound_entry_confidence=_float_env("REBOUND_ENTRY_CONFIDENCE", 0.58),
rebound_min_probability=_float_env("REBOUND_MIN_PROBABILITY", 0.58),
rebound_max_position_usdt=_float_env("REBOUND_MAX_POSITION_USDT", 6.0),
kelly_sizing_enabled=_bool_env("KELLY_SIZING_ENABLED", True),
kelly_fraction=_float_env("KELLY_FRACTION", 0.25),
kelly_max_fraction=_float_env("KELLY_MAX_FRACTION", 0.20),
time_series_forecast_enabled=_bool_env("TIME_SERIES_FORECAST_ENABLED", True),
time_series_min_candles=_int_env("TIME_SERIES_MIN_CANDLES", 120),
time_series_validation_window=_int_env("TIME_SERIES_VALIDATION_WINDOW", 30),
crypto_spot_bot/dashboard.py
+4
View File
@@ -209,6 +209,9 @@ def _safe_config(settings: Settings) -> dict[str, Any]:
"rebound_entry_confidence": settings.rebound_entry_confidence,
"rebound_min_probability": settings.rebound_min_probability,
"rebound_max_position_usdt": settings.rebound_max_position_usdt,
"kelly_sizing_enabled": settings.kelly_sizing_enabled,
"kelly_fraction": settings.kelly_fraction,
"kelly_max_fraction": settings.kelly_max_fraction,
"time_series_forecast_enabled": settings.time_series_forecast_enabled,
"time_series_min_candles": settings.time_series_min_candles,
"time_series_validation_window": settings.time_series_validation_window,
@@ -930,6 +933,7 @@ HTML = r"""
['Rebound-режим', yesNo(config.rebound_trading_enabled)],
['Rebound порог / вероятность', `${num(config.rebound_entry_confidence, 2)} / ${num(config.rebound_min_probability, 2)}`],
['Rebound макс. размер', money(config.rebound_max_position_usdt)],
['Kelly размер', `${yesNo(config.kelly_sizing_enabled)} · ${num(config.kelly_fraction, 2)}x · max ${num((config.kelly_max_fraction || 0) * 100, 1)}%`],
['Прогноз временных рядов', yesNo(config.time_series_forecast_enabled)],
['Модельный горизонт', `${config.time_series_forecast_horizon} свечи`],
['Walk-forward окно', `${config.time_series_validation_window} свечей`],
crypto_spot_bot/strategy.py
+92 -17
View File
@@ -21,6 +21,7 @@ class SpotStrategy:
learning: dict | None = None,
llm: dict | None = None,
forecast: dict | None = None,
account: dict | None = None,
) -> Signal:
if ticker is None:
return Signal(symbol, "HOLD", 0.0, "нет ticker-данных")
@@ -134,15 +135,16 @@ class SpotStrategy:
pattern_blocks_entry = negative_pattern and not (
rebound["active"] and rebound_entry_score >= entry_threshold
)
position_notional = _position_notional(
position_sizing = _position_sizing(
settings=self.settings,
final_score=final_score,
grid_active=grid["active"],
rebound_active=rebound["active"],
llm=llm,
forecast=forecast,
adaptive=adaptive,
account=account,
)
position_notional = float(position_sizing["notional_usdt"])
trade_mode = "GRID" if grid["active"] else "REBOUND" if rebound["active"] else "NORMAL"
diagnostics = {
@@ -163,6 +165,7 @@ class SpotStrategy:
"falling_market": falling_market,
"open_positions_for_symbol": open_positions_for_symbol,
"position_notional_usdt": position_notional,
"position_sizing": position_sizing,
"trade_mode": trade_mode,
"base_entry_threshold": round(base_entry_threshold, 4),
"adaptive_entry_threshold_adjustment": round(adaptive_entry_adjustment, 4),
@@ -473,16 +476,16 @@ def _clamp(value: float, low: float, high: float) -> float:
return max(low, min(high, value))
def _position_notional(
def _position_sizing(
*,
settings: Settings,
final_score: float,
grid_active: bool,
rebound_active: bool,
llm: dict,
forecast: dict | None = None,
adaptive: dict | None = None,
) -> float:
account: dict | None = None,
) -> dict[str, float | bool | str]:
low = max(0.0, settings.min_position_usdt)
high = max(low, settings.max_position_usdt)
if grid_active:
@@ -491,28 +494,100 @@ def _position_notional(
high = max(low, min(high, settings.rebound_max_position_usdt))
denominator = max(0.0001, 1.0 - settings.min_signal_confidence)
confidence_ratio = _clamp((final_score - settings.min_signal_confidence) / denominator, 0.0, 1.0)
raw = low + (high - low) * confidence_ratio
risk = str(llm.get("risk_level", "medium")).lower()
if risk == "high":
raw *= 0.55
elif risk == "low":
raw *= 1.10
confidence_notional = low + (high - low) * confidence_ratio
risk_multiplier = _position_risk_multiplier(forecast, adaptive)
method = "confidence"
raw = confidence_notional
kelly = _kelly_position(
settings=settings,
final_score=final_score,
forecast=forecast or {},
adaptive=adaptive or {},
account=account,
)
if settings.kelly_sizing_enabled:
method = "fractional_kelly"
raw = float(kelly["kelly_notional_usdt"])
raw *= risk_multiplier
notional = round(_clamp(raw, low, high), 2)
return {
"method": method,
"enabled": bool(settings.kelly_sizing_enabled),
"notional_usdt": notional,
"confidence_notional_usdt": round(confidence_notional, 2),
"risk_multiplier": round(risk_multiplier, 4),
"low_cap_usdt": round(low, 2),
"high_cap_usdt": round(high, 2),
**kelly,
}
def _position_risk_multiplier(forecast: dict | None, adaptive: dict | None) -> float:
multiplier = 1.0
forecast = forecast or {}
if forecast.get("usable"):
probability_up = _safe_float(forecast.get("probability_up"), 0.5)
volatility_percent = _safe_float(forecast.get("volatility_percent"), 0.0)
if probability_up < 0.52:
raw *= 0.75
multiplier *= 0.75
elif probability_up >= 0.60:
raw *= 1.08
multiplier *= 1.08
if volatility_percent >= 0.8:
raw *= 0.70
multiplier *= 0.70
risk_mode = str((adaptive or {}).get("risk_mode", "neutral")).lower()
if risk_mode == "defensive":
raw *= 0.65
multiplier *= 0.65
elif risk_mode == "expansion":
raw *= 1.10
return round(_clamp(raw, low, high), 2)
multiplier *= 1.10
return multiplier
def _kelly_position(
*,
settings: Settings,
final_score: float,
forecast: dict,
adaptive: dict,
account: dict | None,
) -> dict[str, float | bool | str]:
confidence_probability = _confidence_probability(final_score, settings.min_signal_confidence)
probability_source = "confidence"
probability = confidence_probability
if forecast.get("usable"):
probability = _safe_float(forecast.get("probability_up"), confidence_probability)
probability_source = "forecast"
probability = _clamp(probability, 0.0, 1.0)
stop_loss = _adaptive_percent(adaptive, "stop_loss_percent", settings.stop_loss_percent, 0.003, 0.08)
take_profit = _adaptive_percent(adaptive, "take_profit_percent", settings.take_profit_percent, 0.003, 0.20)
round_trip_cost = max(0.0, 2.0 * (settings.taker_fee_rate + settings.slippage_rate))
win_return = max(0.0, take_profit - round_trip_cost)
loss_return = max(0.0001, stop_loss + round_trip_cost)
reward_loss_ratio = win_return / loss_return if loss_return > 0 else 0.0
full_kelly = probability - ((1.0 - probability) / reward_loss_ratio) if reward_loss_ratio > 0 else 0.0
full_kelly = max(0.0, full_kelly)
fractional_kelly = full_kelly * _clamp(settings.kelly_fraction, 0.0, 1.0)
effective_fraction = _clamp(fractional_kelly, 0.0, _clamp(settings.kelly_max_fraction, 0.0, 1.0))
bankroll = _safe_float((account or {}).get("equity"), settings.starting_balance_usdt)
if bankroll <= 0:
bankroll = settings.starting_balance_usdt
kelly_notional = max(0.0, bankroll * effective_fraction)
return {
"kelly_probability": round(probability, 4),
"kelly_probability_source": probability_source,
"kelly_reward_loss_ratio": round(reward_loss_ratio, 4),
"kelly_full_fraction": round(full_kelly, 4),
"kelly_fractional_fraction": round(fractional_kelly, 4),
"kelly_effective_fraction": round(effective_fraction, 4),
"kelly_bankroll_usdt": round(bankroll, 2),
"kelly_notional_usdt": round(kelly_notional, 2),
}
def _confidence_probability(final_score: float, min_signal_confidence: float) -> float:
denominator = max(0.0001, 1.0 - min_signal_confidence)
ratio = _clamp((final_score - min_signal_confidence) / denominator, 0.0, 1.0)
return 0.50 + ratio * 0.18
def _grid_state(