Rotation Plan Calendar
Generated {today.strftime('%Y-%m-%d')} · {N_SLICES} slices of ${CAPITAL:,} (${total_capital:,} in total) · estimated total net ${total_high:,.2f} at {TAX_RATE_HIGH:.0%} tax, ${total_low:,.2f} at {TAX_RATE_LOW:.0%} tax
"""]
blocked = [(s, st) for s, st in sorted((states or {}).items()) if st["blocked"]]
busy = [(s, st) for s, st in sorted((states or {}).items())
if st["blocked"] is None and st["holding"]]
if blocked or busy:
lines = []
for s, st in blocked:
lines.append(f'Slice {s} is blocked: it still holds {st["blocked"]}, '
f'which has no sell date yet. No trades are scheduled for it.')
for s, st in busy:
lines.append(f'Slice {s} holds {st["holding"]} until {str(st["hold_until"])}, '
f'so its first trade below starts from that date.')
parts.append('
' + '
'.join(lines) + '
')
parts.append('
')
for _, row in plan_df.iterrows():
sl = int(row["Slice"])
color = palette[(sl - 1) % len(palette)]
star = "★ " if row.get("Pref") == "Yes" else ""
parts.append(f''
f'{star}S{sl} {row["Ticker"]} (net ${row[COL_NET_HIGH]:,.2f} to ${row[COL_NET_LOW]:,.2f})')
parts.append('
')
for (yy, mm) in months:
parts.append(f'
{cal.month_name[mm]} {yy}'
+ ''.join(f'| {w} | ' for w in ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]) + '
')
for week in cal.Calendar().monthdatescalendar(yy, mm):
parts.append('')
for day in week:
if day.month != mm:
parts.append(' | ')
continue
style = ""
if day in tints:
style = f' style="background:{tints[day]}22;"'
cls = ' class="we"' if day.weekday() >= 5 else ''
cell = f'{day.day}'
for label, color in chips.get(day, []):
cell += f'{label}'
cell += ''
parts.append(cell)
parts.append('
')
parts.append('
')
parts.append('
S1 to S4 = which slice of capital pays for the trade. Each slice holds '
'one position at a time and rotates on its own clock. '
'★ = a preferred name: judged safe to keep holding if the rebound runs long. '
'Tinted days = that slice\'s capital is occupied. '
'Sell days are estimates based on each stock\'s average rebound time. '
f'Net figures show a range: taxed at {TAX_RATE_HIGH:.0%} (conservative) to {TAX_RATE_LOW:.0%} (optimistic).
')
parts.append('
')
with open(out_file, "w", encoding="utf-8") as f:
f.write(''.join(parts))
def save_excel(shown: pd.DataFrame, plan_df: pd.DataFrame, out_file: str):
"""Formatted Excel copy of the results: bold frozen header, sized columns,
centred data (long names stay left-aligned). Two sheets: the ranked
candidates and the rotation plan."""
from openpyxl.styles import Font, PatternFill, Alignment
from openpyxl.utils import get_column_letter
def style_sheet(ws, left_aligned=("Name",)):
header_font = Font(bold=True, color="FFFFFF")
header_fill = PatternFill(start_color="1F4E79", end_color="1F4E79", fill_type="solid")
skip_cols = set()
for cell in ws[1]:
cell.font = header_font
cell.fill = header_fill
cell.alignment = Alignment(horizontal="center")
if cell.value in left_aligned:
skip_cols.add(cell.column)
ws.freeze_panes = "A2"
centre = Alignment(horizontal="center")
for row in ws.iter_rows(min_row=2):
for cell in row:
if cell.column not in skip_cols:
cell.alignment = centre
for col_idx, col_cells in enumerate(ws.columns, start=1):
width = max(len(str(c.value)) if c.value is not None else 0 for c in col_cells)
ws.column_dimensions[get_column_letter(col_idx)].width = min(width + 3, 35)
with pd.ExcelWriter(out_file, engine="openpyxl") as writer:
shown.to_excel(writer, sheet_name="Candidates", index=True, index_label="#")
style_sheet(writer.sheets["Candidates"])
if plan_df is not None and not plan_df.empty:
plan_df.to_excel(writer, sheet_name="Rotation Plan", index=True, index_label="#")
style_sheet(writer.sheets["Rotation Plan"])
def run_scanner():
today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
total_capital = CAPITAL * N_SLICES
print(f"\nDividend Capture Scanner v10 - {today.strftime('%Y-%m-%d')}")
print(f"Looking for ex-dates within the next {EX_DATE_WINDOW_DAYS} days")
print(f"After-tax figures shown at two rates: {TAX_RATE_HIGH:.0%} (conservative) and {TAX_RATE_LOW:.0%} (optimistic)")
fee_note = f", financing fee {FINANCE_FEE:.0%}" if FINANCE_FEE > 0 else ""
print(f"Rotation capital: {N_SLICES} slices of ${CAPITAL:,} = ${total_capital:,}{fee_note}")
if HOLDINGS:
print(f"Currently held ({len(HOLDINGS)}): "
+ ", ".join(f"{h['ticker']} in slice {h['slice']}" for h in HOLDINGS)
+ " - these are excluded from the plan")
if EXCLUDED_SECTORS:
print(f"Excluded sectors: {', '.join(EXCLUDED_SECTORS)}")
print(f"Preferred list: {len(FAVORED_TICKERS)} names flagged and favored in the plan\n")
print("Fetching universe (S&P 500 + high-yield names + CEFs + preferreds)...")
tickers = get_universe()
print(f"Scanning {len(tickers)} stocks with {MAX_WORKERS} parallel workers...")
print("(Stocks that pass the basic filters also get a rebound, trend,")
print(" entry-timing, and recovery analysis from 2 years of history.)\n")
candidates, errors, excluded = [], [], []
done = 0
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
futures = {pool.submit(fetch_candidate, t, today): t for t in tickers}
for future in as_completed(futures):
done += 1
row, note = future.result()
if row:
candidates.append(row)
elif note and note.startswith("EXCLUDED:"):
excluded.append(note[len("EXCLUDED:"):])
elif note:
errors.append(note)
if done % 100 == 0:
print(f" {done}/{len(tickers)} scanned, {len(candidates)} candidates so far...")
if excluded:
print(f"\nExcluded by sector ({len(excluded)}): {', '.join(sorted(excluded))}")
if errors:
print(f"\nNote: {len(errors)} tickers were skipped due to data errors.")
if len(errors) <= 10:
for e in errors:
print(f" - {e}")
if not candidates:
print("\nNo candidates found matching the current filters.")
if len(errors) > len(tickers) * 0.5:
print("More than half the tickers errored - Yahoo Finance may be "
"blocking requests or has changed its data format.")
return
df = pd.DataFrame(candidates)
df["Score"] = df.apply(score, axis=1).round(1)
df = df.sort_values("Score", ascending=False).reset_index(drop=True)
df.index += 1
display_cols = [
"Ticker", "Name", "Pref", "Sector", "Price", "Ex-Date", "Days Until",
"Annual Yield", "Capture Yield", "Div/100sh", "Capital",
COL_AT_HIGH, COL_AT_LOW,
"Avg Reb", "Worst Reb", "Recovered", "Best Entry", "Drop Ratio",
"Trend", "Liquidity", "Beta", "Ann.Eff%", "Score",
]
shown = df[display_cols]
print(f"\n{'='*140}")
print(f" TOP DIVIDEND CAPTURE CANDIDATES ({len(df)} found)")
print(f"{'='*140}")
print(shown.to_string())
print(f"\nColumn guide:")
print(f" Pref = on the preferred list: names judged safe to keep holding if a")
print(f" rebound takes far longer than expected; they win close calls")
print(f" in the Rotation Plan (see FAVORED_TICKERS at the top of the script)")
print(f" Sector = the stock's sector. Whole sectors can be ruled out (see")
print(f" EXCLUDED_SECTORS); the plan also warns when slices overlap in one")
print(f" Annual Yield = full year's dividend as % of price (ranked by this, higher is better)")
print(f" Capture Yield = single dividend payment as % of price (what you collect per trade)")
print(f" Div/100sh = dividend dollars for one 100-share round lot")
print(f" Capital = cost of 100 shares")
print(f" {COL_AT_HIGH}= dividend dollars per 100 shares after {TAX_RATE_HIGH:.0%} tax (the conservative estimate)")
print(f" {COL_AT_LOW}= the same after {TAX_RATE_LOW:.0%} tax (the optimistic estimate)")
print(f" Avg/Worst Reb = average and worst-case trading days to recover to the pre-ex price")
print(f" ('>{REBOUND_MAX_DAYS}' means at least one past event never recovered in the window)")
print(f" Recovered = how often it recovered within {REBOUND_MAX_DAYS} trading days")
print(f" Best Entry = back-tested sweet spot: how many trading days before the ex-date")
print(f" buying has historically worked best for this stock")
print(f" Drop Ratio = how much of the dividend the price actually gives up on the ex-date")
print(f" (1.0 = drops the full dividend; below 1.0 = drops less, which is good)")
print(f" Trend = total-return price trend (Up / Down / Mixed). Downtrends are discouraged.")
print(f" Liquidity = Thin / Moderate / High (thin = harder to exit)")
print(f" Ann.Eff% = theoretical annualised after-tax return with continuous rotation,")
print(f" taxed at {TAX_RATE_HIGH:.0%}; failed recoveries are charged at the full {REBOUND_MAX_DAYS}-day wait")
print(f" Score = Ann.Eff% adjusted down for downtrend and thin liquidity")
plan_df, states = build_rotation_plan(df, today)
blocked = [(s, st) for s, st in sorted(states.items()) if st["blocked"]]
busy = [(s, st) for s, st in sorted(states.items())
if st["blocked"] is None and st["holding"]]
if blocked or busy:
print(f"\n{'='*140}")
print(" SLICE STATUS")
print(f"{'='*140}")
for s, st in blocked:
print(f" Slice {s}: BLOCKED - still holds {st['blocked']} with no sell date. "
f"No trades scheduled for this slice.")
for s, st in busy:
print(f" Slice {s}: holds {st['holding']} until {str(st['hold_until'])}. "
f"Its first trade below starts from that date.")
free_now = N_SLICES - len(blocked) - len(busy)
print(f" {free_now} of {N_SLICES} slices are free today "
f"(${free_now * CAPITAL:,} of ${total_capital:,}).")
if not plan_df.empty:
total_high = plan_df[COL_RUN_HIGH].iloc[-1]
total_low = plan_df[COL_RUN_LOW].iloc[-1]
tradeable = (N_SLICES - len(blocked)) * CAPITAL
print(f"\n{'='*140}")
print(f" ROTATION PLAN ({N_SLICES} slices of ${CAPITAL:,}, net shown at "
f"{TAX_RATE_HIGH:.0%} and {TAX_RATE_LOW:.0%} tax"
f"{', ' + format(FINANCE_FEE, '.0%') + ' financing fee' if FINANCE_FEE > 0 else ''})")
print(f"{'='*140}")
print(plan_df.to_string())
print(f"\n {len(plan_df)} trades planned over the next {EX_DATE_WINDOW_DAYS} days "
f"across {plan_df['Slice'].nunique()} slices.")
print(f" Estimated total net: ${total_high:,.2f} at {TAX_RATE_HIGH:.0%} tax "
f"({total_high / tradeable:.2%} of the ${tradeable:,} actually tradeable) to "
f"${total_low:,.2f} at {TAX_RATE_LOW:.0%} tax ({total_low / tradeable:.2%}).")
print(f" Slice = which pot pays for the trade. Buy = the stock's own back-tested Best")
print(f" Entry day. Est. Sell = ex-date plus its average rebound. Cash Free = one")
print(f" settlement day after the sale (T+{SETTLEMENT_DAYS}).")
print(f" Worst Reb shows the honest risk: the longest that stock has ever taken to recover.")
overlaps = sector_overlaps(plan_df)
if overlaps:
print(f"\n SECTOR CONCENTRATION WARNING")
for sec, ticks in overlaps:
print(f" {sec}: {', '.join(ticks)} would be held at the same time.")
print(f" Separate slices are meant to spread risk. Holding one sector in several")
print(f" of them at once turns the whole allocation back into a single bet.")
cal_file = f"rotation_calendar_{today.strftime('%Y%m%d')}.html"
try:
save_calendar_html(plan_df, today, cal_file, states)
print(f"\nSaved to {cal_file} (visual calendar - open it in any browser)")
except PermissionError:
print(f"\nCould not save {cal_file} - close it in the browser/editor and rerun.")
else:
print("\nNo rotation plan could be built. Either no candidates have reliable")
print("rebound history, or every slice is already committed.")
out_file = f"candidates_{today.strftime('%Y%m%d')}.csv"
shown.to_csv(out_file, index=True)
print(f"\nSaved to {out_file}")
out_xlsx = f"candidates_{today.strftime('%Y%m%d')}.xlsx"
try:
save_excel(shown, plan_df, out_xlsx)
print(f"Saved to {out_xlsx} (formatted for Excel, includes the Rotation Plan sheet)")
except PermissionError:
print(f"Could not save {out_xlsx} - close the file if it is open in Excel and rerun.")
except ImportError:
print(f"Skipped {out_xlsx} - install openpyxl first: pip install openpyxl")
if __name__ == "__main__":
run_scanner()