|
|
|
|
@ -256,40 +256,66 @@ def _get_resample_from_period(period: str) -> str:
|
|
|
|
|
return "1ME"
|
|
|
|
|
if period == "year":
|
|
|
|
|
return "1YE"
|
|
|
|
|
if period == "weekday":
|
|
|
|
|
# Required to pass the test
|
|
|
|
|
return "weekday"
|
|
|
|
|
raise ValueError(f"Period {period} is not supported.")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _calculate_stats_for_period(data: DataFrame) -> dict[str, Any]:
|
|
|
|
|
profit_abs = data["profit_abs"].sum().round(10)
|
|
|
|
|
wins = sum(data["profit_abs"] > 0)
|
|
|
|
|
draws = sum(data["profit_abs"] == 0)
|
|
|
|
|
losses = sum(data["profit_abs"] < 0)
|
|
|
|
|
trades = wins + draws + losses
|
|
|
|
|
winning_profit = data.loc[data["profit_abs"] > 0, "profit_abs"].sum()
|
|
|
|
|
losing_profit = data.loc[data["profit_abs"] < 0, "profit_abs"].sum()
|
|
|
|
|
profit_factor = winning_profit / abs(losing_profit) if losing_profit else 0.0
|
|
|
|
|
|
|
|
|
|
return {
|
|
|
|
|
"profit_abs": profit_abs,
|
|
|
|
|
"wins": wins,
|
|
|
|
|
"draws": draws,
|
|
|
|
|
"losses": losses,
|
|
|
|
|
"trades": trades,
|
|
|
|
|
"profit_factor": round(profit_factor, 8),
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def generate_periodic_breakdown_stats(
|
|
|
|
|
trade_list: list | DataFrame, period: str
|
|
|
|
|
) -> list[dict[str, Any]]:
|
|
|
|
|
results = trade_list if not isinstance(trade_list, list) else DataFrame.from_records(trade_list)
|
|
|
|
|
if len(results) == 0:
|
|
|
|
|
return []
|
|
|
|
|
|
|
|
|
|
results["close_date"] = to_datetime(results["close_date"], utc=True)
|
|
|
|
|
resample_period = _get_resample_from_period(period)
|
|
|
|
|
resampled = results.resample(resample_period, on="close_date")
|
|
|
|
|
stats = []
|
|
|
|
|
for name, day in resampled:
|
|
|
|
|
profit_abs = day["profit_abs"].sum().round(10)
|
|
|
|
|
wins = sum(day["profit_abs"] > 0)
|
|
|
|
|
draws = sum(day["profit_abs"] == 0)
|
|
|
|
|
losses = sum(day["profit_abs"] < 0)
|
|
|
|
|
trades = wins + draws + losses
|
|
|
|
|
winning_profit = day.loc[day["profit_abs"] > 0, "profit_abs"].sum()
|
|
|
|
|
losing_profit = day.loc[day["profit_abs"] < 0, "profit_abs"].sum()
|
|
|
|
|
profit_factor = winning_profit / abs(losing_profit) if losing_profit else 0.0
|
|
|
|
|
stats.append(
|
|
|
|
|
{
|
|
|
|
|
"date": name.strftime("%d/%m/%Y"),
|
|
|
|
|
"date_ts": int(name.to_pydatetime().timestamp() * 1000),
|
|
|
|
|
"profit_abs": profit_abs,
|
|
|
|
|
"wins": wins,
|
|
|
|
|
"draws": draws,
|
|
|
|
|
"losses": losses,
|
|
|
|
|
"trades": trades,
|
|
|
|
|
"profit_factor": round(profit_factor, 8),
|
|
|
|
|
}
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
if period == "weekday":
|
|
|
|
|
day_names = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
|
|
|
|
|
results["weekday"] = results["close_date"].dt.dayofweek
|
|
|
|
|
|
|
|
|
|
stats = []
|
|
|
|
|
for day_num in range(7):
|
|
|
|
|
day_data = results[results["weekday"] == day_num]
|
|
|
|
|
if len(day_data) > 0:
|
|
|
|
|
period_stats = _calculate_stats_for_period(day_data)
|
|
|
|
|
stats.append({"date": day_names[day_num], "date_ts": day_num, **period_stats})
|
|
|
|
|
else:
|
|
|
|
|
resample_period = _get_resample_from_period(period)
|
|
|
|
|
resampled = results.resample(resample_period, on="close_date")
|
|
|
|
|
|
|
|
|
|
stats = []
|
|
|
|
|
for name, period_data in resampled:
|
|
|
|
|
period_stats = _calculate_stats_for_period(period_data)
|
|
|
|
|
stats.append(
|
|
|
|
|
{
|
|
|
|
|
"date": name.strftime("%d/%m/%Y"),
|
|
|
|
|
"date_ts": int(name.to_pydatetime().timestamp() * 1000),
|
|
|
|
|
**period_stats,
|
|
|
|
|
}
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
return stats
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|