gitea/historical-prices
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add weekly and bi-weekly frequently

Browse Source
This commit is contained in:
tab888
2026-01-28 12:09:19 +08:00
parent cf708d2466
commit d33b521b22
5 changed files with 431 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
engine.py
+167 -36
View File
@@ -478,32 +478,52 @@ class StrategyEngine:
anchor_day = start_dt_obj.day
latest_csv_date = df['date'].max() # This captures 2026-01-27
# 3. Filter data starting from your start_date
df_filtered = df[df['date'] >= start_dt_obj].copy()
# --- 3. Filter data starting from your start_date ---
df_from_start = df[df['date'] >= start_dt_obj].copy()
if df_from_start.empty:
return []
# 4. Select recurring monthly days (The first trading day on/after the anchor day)
monthly_df = df_filtered[df_filtered['date'].dt.day >= anchor_day].groupby([
df_filtered['date'].dt.year,
df_filtered['date'].dt.month
], as_index=False).first()
# --- 4. Choose rows based on Frequency ---
if frequency == "Weekly":
# Take every 5th trading day
final_df = df_from_start.iloc[::5].copy()
step_increment = float((monthly_goal * 12) / 52)
# 5. FORCE LAST ROW: If the latest date from CSV isn't in our list, append it
if monthly_df.empty or monthly_df.iloc[-1]['date'] != latest_csv_date:
last_row = df_filtered[df_filtered['date'] == latest_csv_date]
monthly_df = pd.concat([monthly_df, last_row]).drop_duplicates(subset=['date'])
elif frequency == "Bi-Weekly":
# Take every 10th trading day
final_df = df_from_start.iloc[::10].copy()
step_increment = float((monthly_goal * 12) / 26)
# 6. Finalize index for the strategy loop
monthly_df.index = pd.to_datetime(monthly_df['date'])
else: # Default to Monthly
# Group by year/month and take the first available day >= anchor_day
final_df = df_from_start[df_from_start['date'].dt.day >= anchor_day].groupby([
df_from_start['date'].dt.year,
df_from_start['date'].dt.month
], as_index=False).first()
step_increment = float(monthly_goal)
if monthly_df.empty:
# --- 5. FORCE LAST ROW SAFETY ---
# Reset index to make concatenation clean
final_df = final_df.reset_index(drop=True)
if not df_from_start.empty:
latest_row = df_from_start.iloc[[-1]] # Gets the absolute last available day
if final_df.empty or latest_row.iloc[0]['date'] != final_df.iloc[-1]['date']:
final_df = pd.concat([final_df, latest_row]).drop_duplicates(subset=['date'])
# Sort and final index reset before the loop
final_df = final_df.sort_values('date').reset_index(drop=True)
if final_df.empty:
return []
# Helper for share calculation based on user toggle
# --- 6. Helper for share calculation ---
def get_shares(cash, prc):
if prc <= 0: return 0
# Note: allow_fractional should be defined in your class/scope
return cash / prc if allow_fractional else math.floor(cash / prc)
# 2. Initial Setup
# --- 7. Initial Setup ---
va_shares = 0
dca_shares = 0
va_invested = 0
@@ -511,41 +531,43 @@ class StrategyEngine:
va_target_value = 0
history = []
# 3. Strategy Loop
for i, row in monthly_df.iterrows():
actual_date_str = i.strftime('%Y-%m-%d')
# --- 8. Strategy Loop ---
for step, (idx, row) in enumerate(final_df.iterrows()):
actual_date_str = row['date'].strftime('%Y-%m-%d')
price = float(row['close'])
if i == monthly_df.index[0]:
# --- MONTH 0: INITIAL DEPOSIT ---
actual_inv = initial_inv # This is the 'va_diff'
if step == 0:
# --- STEP 0: INITIAL DEPOSIT ---
actual_inv = initial_inv
dca_actual_inv = initial_inv
va_target_value = initial_inv
diff = 0
va_new_shares = get_shares(actual_inv, price)
dca_new_shares = va_new_shares
else:
# --- MONTH 1+: DVA vs DCA ---
# --- STEP 1+: DVA vs DCA ---
# Use step_increment (calculated in your frequency logic)
# to ensure growth matches frequency (Weekly/Monthly)
current_increment = step_increment
# DCA Logic
dca_actual_inv = monthly_target
dca_actual_inv = current_increment
dca_new_shares = get_shares(dca_actual_inv, price)
# DVA Logic (Fixed Value Path)
va_target_value += monthly_target
va_target_value += current_increment
# Gap calculation: Target vs. current value BEFORE this month's investment
# Gap calculation: Target vs. current value BEFORE this step's investment
current_va_val_pre = va_shares * price
diff = va_target_value - current_va_val_pre
# Apply Buy/Sell constraints
# note: allow_sell should be defined in your class/scope
actual_inv = diff if (diff >= 0 or allow_sell) else 0
va_new_shares = get_shares(actual_inv, price)
# --- STATE UPDATES (Must happen for both Month 0 and Month 1+) ---
# --- STATE UPDATES ---
va_shares += va_new_shares
dca_shares += dca_new_shares
@@ -553,7 +575,6 @@ class StrategyEngine:
dca_invested += dca_actual_inv
# --- Unified History Append ---
# We calculate these here so they are ALWAYS defined for every row
history.append({
"date": actual_date_str,
"price": round(price, 2),
@@ -561,14 +582,124 @@ class StrategyEngine:
"dca_invested": round(dca_invested, 2),
"dca_shares_trans": round(dca_new_shares, 4),
"dca_shares_total": round(dca_shares, 4),
"va_value": round(va_shares * price, 2), # Becomes 'Current Portfolio Value'
"va_invested": round(va_invested, 2), # Becomes 'Total Invested'
"va_value": round(va_shares * price, 2),
"va_invested": round(va_invested, 2),
"va_diff": round(actual_inv, 2),
"va_shares_trans": round(va_new_shares, 4),
"va_shares_total": round(va_shares, 4),
"va_target_value": round(va_target_value, 2) # Used for next goal
"va_target_value": round(va_target_value, 2)
})
# Debugging print
print(f"Date: {i.strftime('%Y-%m')}, Target: {va_target_value:.2f}, Portfolio: {va_invested:.2f}, Diff: {diff:.2f}")
return history
def run_simulation(self, start_date, monthly_goal, initial_inv, frequency="Monthly", allow_sell=True, allow_fractional=True):
# 1. DATA LOADING
df = pd.read_csv(self.data_engine.file_path)
df['date'] = pd.to_datetime(df['date'])
df = df.sort_values('date')
# 2. DATE PREP
start_dt_obj = pd.to_datetime(start_date)
latest_csv_date = df['date'].max()
# Filter data starting from user's choice
df_from_start = df[df['date'] >= start_dt_obj].copy()
if df_from_start.empty:
return []
# 3. FREQUENCY LOGIC (Hardened)
# Standardize string for comparison
freq_check = str(frequency).strip().title()
if freq_check == "Weekly":
final_df = df_from_start.iloc[::5].copy()
step_increment = float((monthly_goal * 12) / 52)
elif freq_check == "Bi-Weekly":
final_df = df_from_start.iloc[::10].copy()
step_increment = float((monthly_goal * 12) / 26)
else:
# For Monthly, we need the anchor day
anchor_day = start_dt_obj.day
final_df = df_from_start[df_from_start['date'].dt.day >= anchor_day].groupby([
df_from_start['date'].dt.year,
df_from_start['date'].dt.month
], as_index=False).first()
step_increment = float(monthly_goal)
# 4. SAFETY: Ensure most recent price is included
final_df = final_df.reset_index(drop=True)
last_actual_row = df_from_start.iloc[[-1]]
if final_df.empty or last_actual_row.iloc[0]['date'] != final_df.iloc[-1]['date']:
final_df = pd.concat([final_df, last_actual_row]).drop_duplicates(subset=['date'])
final_df = final_df.sort_values('date').reset_index(drop=True)
# 5. STRATEGY INITIALIZATION
def get_shares(cash, prc):
if prc <= 0: return 0
if allow_fractional:
return float(cash / prc)
else:
# Handles both buying (+) and selling (-) for whole shares
return float(math.floor(cash / prc)) if cash >= 0 else float(math.ceil(cash / prc))
# Ensure these are initialized as floats
va_shares, dca_shares = 0.0, 0.0
va_invested, dca_invested = 0.0, 0.0
history = []
for step, (idx, row) in enumerate(final_df.iterrows()):
actual_date_str = row['date'].strftime('%Y-%m-%d')
price = float(row['close'])
if step == 0:
# First row: Both strategies start with the Initial Investment
va_actual_inv = float(initial_inv)
dca_actual_inv = float(initial_inv)
va_target_value = float(initial_inv)
else:
# Subsequent rows: Use the frequency-adjusted step_increment
va_target_value += step_increment
# DCA logic: Always invests the same amount every period
dca_actual_inv = float(step_increment)
# VA logic: Invests enough to hit the target value
current_va_market_val = va_shares * price
diff = va_target_value - current_va_market_val
# Apply "Allow Sell" constraint
va_actual_inv = diff if (diff >= 0 or allow_sell) else 0.0
# Update Shares based on fractional setting
va_new_shares = get_shares(va_actual_inv, price)
dca_new_shares = get_shares(dca_actual_inv, price)
# Running totals for shares
va_shares += va_new_shares
dca_shares += dca_new_shares
# Running totals for principal invested
va_invested += va_actual_inv
dca_invested += dca_actual_inv
history.append({
"date": actual_date_str,
"price": round(price, 2),
"va_diff": round(va_actual_inv, 2), # Invested this step (VA)
"va_shares_trans": round(va_new_shares, 4),
"va_value": round(va_shares * price, 2), # Current Portfolio Value (VA)
"va_invested": round(va_invested, 2), # Total Out-of-Pocket (VA)
"va_shares_total": round(va_shares, 4),
"va_target_value": round(float(va_target_value), 2),
"dca_diff": round(dca_actual_inv, 2), # Invested this step (DCA)
"dca_shares_trans": round(dca_new_shares, 4),
"dca_value": round(dca_shares * price, 2), # Current Portfolio Value (DCA)
"dca_invested": round(dca_invested, 2), # Total Out-of-Pocket (DCA)
"dca_shares_total": round(dca_shares, 4)
})
return history